Changeset 35

trunk/libsipc/include/sipc/sipc.h

r24	r35
53	53	* the sipc_t struct. Caller must call sipc_close to free the memory.
54	54	*/
55		sipc_t sipc_open(const char key, int role, int ipc_type, size_t size);
	55	sipc_t sipc_open(const char key, int role, int ipc_type, size_t size);
56	56
57	57	/* Free the sipc struct and its contents */
58		void sipc_close(sipc_t *sipc);
	58	void sipc_close(sipc_t * sipc);
59	59
60	60	/* Called by a helper application to destroy an IPC resource.
…	…
65	65	* This can only be called if sender was specified when sipc_init was called.
66	66	* returns 0 on success, <0 on failure */
67		int sipc_send_data(sipc_t *sipc, int msg_len);
	67	int sipc_send_data(sipc_t * sipc, int msg_len);
68	68
69	69	/* Blocking call to recieve data. Data will be allocated and filled and
70	70	* len will be set to the length. Returns 0 on success, <0 on failure */
71		int sipc_recv_data(sipc_t sipc, char data, int len);
	71	int sipc_recv_data(sipc_t * sipc, char *data, int len);
72	72
73	73	/* Returns a pointer to the data contained within the IPC resource */
74		char sipc_get_data_ptr(sipc_t sipc);
	74	char sipc_get_data_ptr(sipc_t sipc);
75	75
76	76	/* Prints an error message, accepts printf format string */
77		void sipc_error(sipc_t sipc, const char fmt, ...)
78		__attribute__ ((format (printf, 2, 3)));
	77	void sipc_error(sipc_t * sipc, const char *fmt, ...)
	78	__attribute__ ((format(printf, 2, 3)));
79	79
80	80	/* Some IPC types unfortunatly need limited access to underlying mechanism */
…	…
82	82
83	83	/* Receiver calls this when it is done receiving a message in shared memory */
84		int sipc_shm_recv_done(sipc_t *sipc);
	84	int sipc_shm_recv_done(sipc_t * sipc);
85	85
86	86	#endif

trunk/libsipc/src/mqueue_internal.c

r32	r35
41	41	int msqid;
42	42
43		msqid = msgget(key, MQ_PERMS~~\|IPC_CREAT\|~~IPC_EXCL);
	43	msqid = msgget(key, MQ_PERMS \| IPC_CREAT \| IPC_EXCL);
44	44	if (msqid < 0)
45	45	fprintf(stderr, "msgget: %s\n", strerror(errno));

trunk/libsipc/src/mqueue_internal.h

r1	r35
19	19	*/
20	20
21
22	21	#ifndef _MQUEUE_INTERNAL_H_
23	22	#define _MQUEUE_INTERNAL_H_

trunk/libsipc/src/shm_internal.c

r30	r35
31	31
32	32	/* Receive NOP messages, which will unblock the sender */
33		int sipc_shm_recv_done(sipc_t *sipc)
	33	int sipc_shm_recv_done(sipc_t * sipc)
34	34	{
35	35	if (!sipc)
…	…
47	47	{
48	48	int shmid = 0;
49
	49
50	50	/* Get the segment's ID */
51	51	shmid = shmget(sipc_key, 0, 0);
…	…
54	54	return;
55	55	}
56
	56
57	57	/* Mark the segment for removal; this will actually
58	58	destroy the segment only if the sender and receiver have detached */
59		if (shmctl(shmid, IPC_RMID, NULL) < 0)
	59	if (shmctl(shmid, IPC_RMID, NULL) < 0)
60	60	fprintf(stderr, "shmctl: %s\n", strerror(errno));
61
	61
62	62	}

trunk/libsipc/src/shm_internal.h

r1	r35
19	19	*/
20	20
21
22	21	#ifndef _SHM_INTERNAL_H_
23	22	#define _SHM_INTERNAL_H_

trunk/libsipc/src/sipc.c

r24	r35
29	29	#include "sipc_internal.h"
30	30
31		int (sipc_init_f[SIPC_NUM_TYPES]) (sipc_t sipc) = {
32		sipc_shm_init,
33		sipc_mqueue_init,
34		};
	31	int (sipc_init_f[SIPC_NUM_TYPES]) (sipc_t sipc) = {
	32	sipc_shm_init, sipc_mqueue_init,};
35	33
36	34	sipc_t sipc_open(const char key, int role, int ipc_type, size_t len)
…	…
59	57
60	58	/* call backend specific init to fill in function table */
61		if (sipc_init_f[ipc_type](new_sipc))
	59	if (sipc_init_f[ipc_type] (new_sipc))
62	60	goto err;
63	61
…	…
81	79
82	80	return new_sipc;
83		err:
	81	err:
84	82	free(new_sipc);
85	83	return NULL;
86	84	}
87	85
88		void sipc_close(sipc_t *sipc)
	86	void sipc_close(sipc_t * sipc)
89	87	{
90	88	if (!sipc)
91	89	return;
92	90
93		sipc->funcs->sipc_detach(sipc);
	91	sipc->funcs->sipc_detach(sipc);
94	92	free(sipc);
95	93	}
…	…
118	116	}
119	117
120		char sipc_get_data_ptr(sipc_t sipc)
	118	char sipc_get_data_ptr(sipc_t sipc)
121	119	{
122	120	if (!sipc)
…	…
126	124	}
127	125
128		int sipc_send_data(sipc_t *sipc, int msg_len)
	126	int sipc_send_data(sipc_t * sipc, int msg_len)
129	127	{
130	128	if (!sipc)
…	…
134	132	}
135	133
136		int sipc_recv_data(sipc_t sipc, char data, int len)
	134	int sipc_recv_data(sipc_t * sipc, char *data, int len)
137	135	{
138	136	if (!sipc)
…	…
142	140	}
143	141
144		void sipc_error(sipc_t sipc, const char fmt, ...)
	142	void sipc_error(sipc_t * sipc, const char *fmt, ...)
145	143	{
146	144	if (!sipc)

trunk/libsipc/src/sipc_internal.h

r21	r35
38	38
39	39	/* Creation functions */
40		int sipc_mqueue_create(sipc_t *sipc);
41		int sipc_shm_create(sipc_t *sipc);
	40	int sipc_mqueue_create(sipc_t * sipc);
	41	int sipc_shm_create(sipc_t * sipc);
42	42	int mqueue_create(key_t key);
43	43
44	44	/* Attach functions */
45		int sipc_mqueue_attach(sipc_t *sipc);
46		int sipc_shm_attach(sipc_t *sipc);
	45	int sipc_mqueue_attach(sipc_t * sipc);
	46	int sipc_shm_attach(sipc_t * sipc);
47	47
48	48	/* Detach functions */
49		void sipc_mqueue_detach(sipc_t *sipc);
50		void sipc_shm_detach(sipc_t *sipc);
	49	void sipc_mqueue_detach(sipc_t * sipc);
	50	void sipc_shm_detach(sipc_t * sipc);
51	51
52	52	/* Destruction functions */
…	…
55	55
56	56	/* Initialization functions */
57		int sipc_mqueue_init(sipc_t *sipc);
58		int sipc_shm_init(sipc_t *sipc);
	57	int sipc_mqueue_init(sipc_t * sipc);
	58	int sipc_shm_init(sipc_t * sipc);
59	59
60	60	/* all ipc backends must implement all of these functions */
61		struct sipc_func_table {
62		int (sipc_create) (sipc_t sipc);
63		int (sipc_attach) (sipc_t sipc);
64		char (sipc_get_data_ptr) (sipc_t *sipc);
65		int (sipc_send_data) (sipc_t sipc, int msg_len);
66		int (sipc_recv_data) (sipc_t sipc, char *data, int len);
67		void (sipc_detach) (sipc_t sipc);
	61	struct sipc_func_table
	62	{
	63	int (sipc_create) (sipc_t sipc);
	64	int (sipc_attach) (sipc_t sipc);
	65	char (sipc_get_data_ptr) (sipc_t * sipc);
	66	int (sipc_send_data) (sipc_t sipc, int msg_len);
	67	int (sipc_recv_data) (sipc_t sipc, char *data, int len);
	68	void (sipc_detach) (sipc_t sipc);
68	69	void (_sipc_error) (const char fmt, va_list ap);
69	70	};
70	71
71		struct sipc {
	72	struct sipc
	73	{
72	74	key_t key;
73	75	int ipc_type;
74	76	int role;
75		union {
76		int fd; /* file desciptor for socket connections */
77		struct shm { /* shm data */
78		int shmid; /* ID of segment */
79		int msqid; /* notify channel */
	77	union
	78	{
	79	int fd; /* file desciptor for socket connections */
	80	struct shm
	81	{ /* shm data */
	82	int shmid; /* ID of segment */
	83	int msqid; /* notify channel */
80	84	/* sipc_t mqueue; handle to notify channel /
81	85	} s;
82		int msqid; /* message queue */
	86	int msqid; /* message queue */
83	87	};
84	88
85	89	char *data;
86		size_t len;
87		size_t msg_len; /* Length in bytes of the next message */
88		size_t copied; /* Number of bytes xmitted so far */
	90	size_t len;
	91	size_t msg_len; /* Length in bytes of the next message */
	92	size_t copied; /* Number of bytes xmitted so far */
89	93	struct sipc_func_table *funcs;
90		};
	94	};

trunk/libsipc/src/sipc_mqueue.c

r34	r35
50	50
51	51	/* Forward decls */
52		static int get_max_msg(sipc_t *sipc);
53		static int mqueue_send_msg_len(sipc_t *sipc, int msg_len);
54		static int mqueue_send_end_xmit(sipc_t *sipc);
55		static char split_data(sipc_t sipc, size_t packet_sz);
	52	static int get_max_msg(sipc_t * sipc);
	53	static int mqueue_send_msg_len(sipc_t * sipc, int msg_len);
	54	static int mqueue_send_end_xmit(sipc_t * sipc);
	55	static char split_data(sipc_t sipc, size_t packet_sz);
56	56	static int is_msg_len(struct msgbuf *mbuf);
57	57	static int is_end_xmit(struct msgbuf *mbuf);
58		static int msg_done(sipc_t *sipc);
	58	static int msg_done(sipc_t * sipc);
59	59	static size_t next_packet_sz(int recv, int max_packet_sz, int msg_len);
60	60	static size_t next_alloc_sz(int recv, int recv_sz, int max_packet_sz, int msg_len);
61	61
62		int sipc_mqueue_init(sipc_t *sipc)
	62	int sipc_mqueue_init(sipc_t * sipc)
63	63	{
64	64	if (!sipc)
…	…
79	79	* This function should only be called by a helper application
80	80	* and not by the sender or reciever. */
81		int sipc_mqueue_create(sipc_t *sipc)
	81	int sipc_mqueue_create(sipc_t * sipc)
82	82	{
83	83	int msqid;
84	84
85	85	msqid = mqueue_create(sipc->key);
86		if (msqid < 0) {
	86	if (msqid < 0) {
87	87	fprintf(stderr, "msgget: %s\n", strerror(errno));
88	88	return -1;
…	…
93	93	}
94	94
95		void sipc_mqueue_detach(sipc_t *sipc)
	95	void sipc_mqueue_detach(sipc_t * sipc)
96	96	{
97	97	free(sipc->data);
…	…
100	100
101	101	/* Attach to an existing message queue. */
102		int sipc_mqueue_attach(sipc_t *sipc)
	102	int sipc_mqueue_attach(sipc_t * sipc)
103	103	{
104	104	/* Allocate data member */
105	105	sipc->data = calloc(1, sipc->len);
106	106	if (!sipc->data) {
107		sipc_error(sipc, "Out of memory\n");
108		return -1;
	107	sipc_error(sipc, "Out of memory\n");
	108	return -1;
109	109	}
110	110
…	…
120	120	}
121	121
122		char sipc_mqueue_get_data_ptr(sipc_t sipc)
	122	char sipc_mqueue_get_data_ptr(sipc_t sipc)
123	123	{
124	124	return sipc->data;
…	…
127	127	/* Sends a message to the queue.
128	128	* Returns 0 on success, -1 on error */
129		int sipc_mqueue_send_data(sipc_t *sipc, int msg_len)
	129	int sipc_mqueue_send_data(sipc_t * sipc, int msg_len)
130	130	{
131	131	int max_packet_sz = SIPC_MQUEUE_MSG_SZ - sizeof(struct msgbuf);
…	…
137	137	goto err;
138	138
139		mbuf = (struct msgbuf *) calloc(1, SIPC_MQUEUE_MSG_SZ);
	139	mbuf = (struct msgbuf *)calloc(1, SIPC_MQUEUE_MSG_SZ);
140	140	if (!mbuf) {
141	141	sipc_error(sipc, "%s\n", "Out of memory");
…	…
146	146	mqueue_send_msg_len(sipc, msg_len);
147	147
148		sipc->copied = 0; /* Reset copied data counter */
	148	sipc->copied = 0; /* Reset copied data counter */
149	149	do {
150	150	packet_sz = next_packet_sz(sipc->copied, max_packet_sz, msg_len);
…	…
173	173	return 0;
174	174
175		err:
	175	err:
176	176	free(packet);
177	177	free(mbuf);
…	…
180	180	}
181	181
182		int sipc_mqueue_recv_data(sipc_t sipc, char data, int len)
	182	int sipc_mqueue_recv_data(sipc_t * sipc, char *data, int len)
183	183	{
184	184	int retv = -1, alloc_sz = 0, idx = 0, recv_sz;
…	…
199	199	}
200	200
201		data = NULL; / Pointer to a buffer created here */
202		idx = 0; /* Current index into the buffer */
203		len = 0; / Bytes received so far */
	201	data = NULL; / Pointer to a buffer created here */
	202	idx = 0; /* Current index into the buffer */
	203	len = 0; / Bytes received so far */
204	204
205	205	/* Receive and validate the length of message marker */
…	…
220	220	if (sipc->msg_len > sipc->len) {
221	221	sipc_error(sipc, "libsipc: cannot receive buffer of size %zd, can only receive %zd\n",
222		sipc->msg_len, sipc->len);
	222	sipc->msg_len, sipc->len);
223	223	goto err;
224	224	}
…	…
240	240	while (!is_end_xmit(mbuf)) {
241	241	/* If this isn't the last packet in the message,
242		* resize to +1 message */
	242	* resize to +1 message */
243	243	alloc_sz += next_alloc_sz(*len, recv_sz, msgtxt_sz, sipc->msg_len);
244	244
…	…
253	253
254	254	/* memset(data+idx, 0x0, recv_sz); /
255		memcpy(*data+idx, mbuf->mtext, recv_sz);
	255	memcpy(*data + idx, mbuf->mtext, recv_sz);
256	256	idx += recv_sz;
257	257
…	…
270	270
271	271	retv = 0;
272		err:
	272	err:
273	273	sipc->msg_len = SIPC_MSGLEN_NOT_SET; /* Unset message length */
274	274	free(mbuf);
…	…
283	283	/* Gets the maximum allowed size of a message in a message queue.
284	284	* Returns the maximum allowed size of a message, -1 on error */
285		static int get_max_msg(sipc_t *sipc)
	285	static int get_max_msg(sipc_t * sipc)
286	286	{
287	287	FILE *f;
…	…
298	298	}
299	299	if (fclose(f) == EOF)
300		sipc_error(sipc, "fclose: %s\n", strerror(errno));
	300	sipc_error(sipc, "fclose: %s\n", strerror(errno));
301	301
302	302	return atoi(buf);
…	…
304	304
305	305	/* Send end of transmission marker */
306		static int mqueue_send_end_xmit(sipc_t *sipc)
	306	static int mqueue_send_end_xmit(sipc_t * sipc)
307	307	{
308	308	struct msgbuf mbuf;
…	…
322	322	{
323	323	int i = 1;
324		while(x) {
	324	while (x) {
325	325	i++;
326	326	x /= 10;
…	…
329	329	}
330	330
331		static int mqueue_send_msg_len(sipc_t *sipc, int len)
332		{
333		struct msgbuf *mbuf;
	331	static int mqueue_send_msg_len(sipc_t * sipc, int len)
	332	{
	333	struct msgbuf *mbuf;
334	334
335	335	if (len < 0)
…	…
339	339	mbuf = calloc(1, sizeof(struct msgbuf) + qlog(len));
340	340	if (!mbuf) {
341		sipc_error(sipc,"Out of memory!\n");
	341	sipc_error(sipc, "Out of memory!\n");
342	342	return -1;
343	343	}
…	…
376	376	/* Returns 1 if the message has been completely fragmented,
377	377	* 0 otherwise. */
378		static int msg_done(sipc_t *sipc)
	378	static int msg_done(sipc_t * sipc)
379	379	{
380	380	return sipc->copied >= sipc->msg_len ? 1 : 0;
381	381	}
382
383	382
384	383	/* Splits data in sipc into a chunk the size of the maximum
…	…
386	385	* This function allocates space for the fragmented piece of data.
387	386	* The caller is responsible for freeign this memory. */
388		static char split_data(sipc_t sipc, size_t packet_sz)
	387	static char split_data(sipc_t sipc, size_t packet_sz)
389	388	{
390	389	size_t remaining = sipc->msg_len - sipc->copied;
…	…
394	393	return NULL;
395	394
396		memcpy(packet, (sipc->data)+(sipc->copied), len);
	395	memcpy(packet, (sipc->data) + (sipc->copied), len);
397	396	sipc->copied += len;
398	397

trunk/libsipc/src/sipc_mqueue.h

r21	r35
28	28	#define MQ_NOBLOCK 0
29	29
30		int sipc_mqueue_init(sipc_t *sipc);
31		int sipc_mqueue_connect(sipc_t *sipc);
32		void sipc_mqueue_disconnect(sipc_t *sipc);
33		char sipc_mqueue_get_data_ptr(sipc_t sipc);
34		int sipc_mqueue_send_data(sipc_t *sipc, int msg_len);
35		int sipc_mqueue_recv_data(sipc_t sipc, char data, int len);
36		int sipc_mqueue_end_xmit(sipc_t *sipc);
37		void sipc_mqueue_destroy_handle(sipc_t *sipc);
	30	int sipc_mqueue_init(sipc_t * sipc);
	31	int sipc_mqueue_connect(sipc_t * sipc);
	32	void sipc_mqueue_disconnect(sipc_t * sipc);
	33	char sipc_mqueue_get_data_ptr(sipc_t sipc);
	34	int sipc_mqueue_send_data(sipc_t * sipc, int msg_len);
	35	int sipc_mqueue_recv_data(sipc_t * sipc, char *data, int len);
	36	int sipc_mqueue_end_xmit(sipc_t * sipc);
	37	void sipc_mqueue_destroy_handle(sipc_t * sipc);
38	38	void sipc_mqueue_error(const char *fmt, va_list ap);
39	39

trunk/libsipc/src/sipc_shm.c

r34	r35
43	43	.sipc_create = sipc_shm_create,
44	44	.sipc_attach = sipc_shm_attach,
45		.sipc_get_data_ptr = sipc_shm_get_data_ptr,
	45	.sipc_get_data_ptr = sipc_shm_get_data_ptr,
46	46	.sipc_send_data = sipc_shm_send_data,
47		.sipc_recv_data = sipc_shm_recv_data,
	47	.sipc_recv_data = sipc_shm_recv_data,
48	48	.sipc_detach = sipc_shm_detach,
49	49	._sipc_error = sipc_shm_error
50	50	};
51	51
52		int sipc_shm_init(sipc_t *sipc)
53		{
54		if (!sipc)
55		return -1;
56
57		sipc->funcs = &shm_funcs;
	52	int sipc_shm_init(sipc_t * sipc)
	53	{
	54	if (!sipc)
	55	return -1;
	56
	57	sipc->funcs = &shm_funcs;
58	58	sipc->data = NULL;
59
60		return 0;
61		}
62
63		int sipc_shm_create(sipc_t *sipc)
	59
	60	return 0;
	61	}
	62
	63	int sipc_shm_create(sipc_t * sipc)
64	64	{
65	65	if (!sipc)
…	…
79	79
80	80	/* Set capacity of side channel to 1 message */
81		if (sipc->role == SIPC_SENDER &&
82		mqueue_set_capacity(sipc->s.msqid, sizeof(struct msgbuf)) < 0)
83		goto err;
84
85		return 0;
86
87		err:
88		if (sipc->s.shmid != (key_t)-1)
	81	if (sipc->role == SIPC_SENDER && mqueue_set_capacity(sipc->s.msqid, sizeof(struct msgbuf)) < 0)
	82	goto err;
	83
	84	return 0;
	85
	86	err:
	87	if (sipc->s.shmid != (key_t) - 1)
89	88	shmctl(sipc->s.shmid, IPC_RMID, NULL);
90	89	return -1;
91	90	}
92	91
93		int sipc_shm_attach(sipc_t *sipc)
	92	int sipc_shm_attach(sipc_t * sipc)
94	93	{
95	94	int flags = 0;
…	…
105	104	goto err;
106	105	}
107
	106
108	107	/* Attach to the shm segment */
109	108	flags = sipc->role == SIPC_SENDER ? 0 : SHM_RDONLY;
110		if ((sipc->data = shmat(sipc->s.shmid, NULL, flags)) == (char *)-1) {
111		sipc_error(sipc, "shmat: %s\n", strerror(errno));
112		goto err;
113		}
	109	if ((sipc->data = shmat(sipc->s.shmid, NULL, flags)) == (char *)-1) {
	110	sipc_error(sipc, "shmat: %s\n", strerror(errno));
	111	goto err;
	112	}
114	113
115	114	/* Connect side channel */
…	…
121	120
122	121	return 0;
123		err:
	122	err:
124	123	sipc_shm_detach(sipc);
125	124	return -1;
126	125	}
127	126
128		void sipc_shm_detach(sipc_t *sipc)
	127	void sipc_shm_detach(sipc_t * sipc)
129	128	{
130	129	if (!sipc)
131	130	return;
132
	131
133	132	/* If we're the receiver, disconnect side channel */
134		if (sipc->role == SIPC_RECEIVER)
	133	if (sipc->role == SIPC_RECEIVER)
135	134	mqueue_disconnect(sipc->s.msqid);
136
	135
137	136	/* Detach shm segment */
138	137	if (sipc->data) {
139		if (shmdt(sipc->data) < 0)
	138	if (shmdt(sipc->data) < 0)
140	139	sipc_error(sipc, "shmdt: %s\n", strerror(errno));
141	140	}
…	…
144	143	void sipc_shm_error(const char *fmt, va_list ap)
145	144	{
146		vfprintf(stderr, fmt, ap);
147		}
148
149		char sipc_shm_get_data_ptr(sipc_t sipc)
	145	vfprintf(stderr, fmt, ap);
	146	}
	147
	148	char sipc_shm_get_data_ptr(sipc_t sipc)
150	149	{
151	150	return sipc->data;
…	…
153	152
154	153	/* Sends a DATA_READY marker, followed by 2 NOP markers */
155		int sipc_shm_send_data(sipc_t *sipc, int msg_len)
156		{
	154	int sipc_shm_send_data(sipc_t * sipc, int msg_len)
	155	{
157	156	if (!sipc)
158	157	return -1;
159	158
160	159	/* Send a DATA_READY marker */
161		if (mqueue_send_msg_type(sipc->s.msqid, SIPC_DATA_READY, MQ_BLOCK) < 0) {
	160	if (mqueue_send_msg_type(sipc->s.msqid, SIPC_DATA_READY, MQ_BLOCK) < 0) {
162	161	sipc_error(sipc, "Could not send DATA_READY marker\n");
163	162	return -1;
164	163	}
165
	164
166	165	/* Send a NULL marker; this should block at first */
167	166	if (mqueue_send_msg_type(sipc->s.msqid, SIPC_NOP, MQ_BLOCK) < 0) {
168	167	sipc_error(sipc, "Could not send NOP marker\n");
169	168	return -1;
170		}
	169	}
171	170
172	171	/* Send another NULL marker */
173	172	if (mqueue_send_msg_type(sipc->s.msqid, SIPC_NOP, MQ_BLOCK) < 0) {
174		sipc_error(sipc, "Could not send NOP marker\n");
175		return -1;
176		}
177
	173	sipc_error(sipc, "Could not send NOP marker\n");
	174	return -1;
	175	}
	176
178	177	return 0;
179	178	}
…	…
183	182	* This function will block if the DATA_READY marker is not
184	183	* present on the queue. */
185		int sipc_shm_recv_data(sipc_t sipc, char data, int len)
186		{
187		int mtype = 0;
188
	184	int sipc_shm_recv_data(sipc_t * sipc, char *data, int len)
	185	{
	186	int mtype = 0;
	187
189	188	if (!sipc \|\| !data \|\| !len)
190	189	return -1;
191	190
192	191	/* Get a message from the side channel; exit when it's END_XMIT */
193		mtype = mqueue_get_msg_type(sipc->s.msqid, SIPC_ANY, MQ_BLOCK);
	192	mtype = mqueue_get_msg_type(sipc->s.msqid, SIPC_ANY, MQ_BLOCK);
194	193	if (mtype == SIPC_DATA_READY) {
195	194	/* It is now OK to read shared memory */
…	…
199	198	} else {
200	199	sipc_error(sipc, "Received a message of unknown type\n");
201		return -1;
202		}
203
204		return 0;
205		}
206
	200	return -1;
	201	}
	202
	203	return 0;
	204	}

trunk/libsipc/src/sipc_shm.h

r21	r35
25	25	#include <stdarg.h>
26	26
27		int sipc_shm_init(sipc_t *sipc);
28		int sipc_shm_connect(sipc_t *sipc);
29		void sipc_shm_disconnect(sipc_t *sipc);
30		char sipc_shm_get_data_ptr(sipc_t sipc);
31		int sipc_shm_send_data(sipc_t *sipc, int msg_len);
32		int sipc_shm_recv_data(sipc_t sipc, char data, int len);
33		int sipc_shm_end_xmit(sipc_t *sipc);
34		void sipc_shm_destroy_handle(sipc_t *sipc);
	27	int sipc_shm_init(sipc_t * sipc);
	28	int sipc_shm_connect(sipc_t * sipc);
	29	void sipc_shm_disconnect(sipc_t * sipc);
	30	char sipc_shm_get_data_ptr(sipc_t sipc);
	31	int sipc_shm_send_data(sipc_t * sipc, int msg_len);
	32	int sipc_shm_recv_data(sipc_t * sipc, char *data, int len);
	33	int sipc_shm_end_xmit(sipc_t * sipc);
	34	void sipc_shm_destroy_handle(sipc_t * sipc);
35	35	void sipc_shm_error(const char *fmt, va_list ap);
36	36

trunk/libsipc/tests/ipc_creator.c

r21	r35
33	33	return -1;
34	34	}
35
	35
36	36	sipc = sipc_open(argv[1], SIPC_CREATOR, atoi(argv[2]), DATA_LEN);
37	37	if (sipc == NULL) {

trunk/libsipc/tests/mqueue.c

r34	r35
41	41	static sipc_t *writer_ipc;
42	42	static sipc_t *reader_ipc;
43		static int send_end_xmit(sipc_t *ipc);
	43	static int send_end_xmit(sipc_t * ipc);
44	44	static void do_parent(void);
45	45	static void do_child(void);
…	…
75	75
76	76	/* Open IPCs */
77		writer_ipc =
78		sipc_open(SIPC_MQ_KEY, SIPC_SENDER, SIPC_SYSV_MQUEUES, DATA_LEN);
	77	writer_ipc = sipc_open(SIPC_MQ_KEY, SIPC_SENDER, SIPC_SYSV_MQUEUES, DATA_LEN);
79	78	if (writer_ipc == NULL) {
80	79	fprintf(stderr, "Unable to initialize IPC resource\n");
81	80	return -1;
82	81	}
83		reader_ipc =
84		sipc_open(SIPC_MQ_KEY, SIPC_RECEIVER, SIPC_SYSV_MQUEUES, DATA_LEN);
	82	reader_ipc = sipc_open(SIPC_MQ_KEY, SIPC_RECEIVER, SIPC_SYSV_MQUEUES, DATA_LEN);
85	83	if (reader_ipc == NULL) {
86	84	fprintf(stderr, "Error initializing IPC resource\n");
…	…
127	125	}
128	126
129		static void test_unforked_mqueue(void) {
	127	static void test_unforked_mqueue(void)
	128	{
130	129	static char message[] = "HELLO,";
131	130	size_t message_len = sizeof(message);
…	…
166	165
167	166	CU_TestInfo mqueue_tests[] = {
168		{"unforked mqueue", test_unforked_mqueue},
169		{"forked mqueue", test_mqueue}
170		,
171		CU_TEST_INFO_NULL
	167	{"unforked mqueue", test_unforked_mqueue}
	168	,
	169	{"forked mqueue", test_mqueue}
	170	,
	171	CU_TEST_INFO_NULL
172	172	};
173	173
…	…
219	219	}
220	220
221		static int send_end_xmit(sipc_t *ipc)
	221	static int send_end_xmit(sipc_t * ipc)
222	222	{
223	223	char *data = sipc_get_data_ptr(ipc);
…	…
228	228
229	229	bzero(data, DATA_LEN);
230		strncpy(data, XMIT_END, DATA_LEN-1);
	230	strncpy(data, XMIT_END, DATA_LEN - 1);
231	231	if (sipc_send_data(ipc, DATA_LEN) < 0) {
232	232	sipc_error(ipc, "Unable to send end of transmission marker\n");

trunk/libsipc/tests/shm.c

r33	r35
46	46	static void do_binary_child(void);
47	47	static void do_binary_parent(void);
48		static void send_end_xmit(sipc_t *sipc);
	48	static void send_end_xmit(sipc_t * sipc);
49	49	static int verify_data(char filename, char recv_data);
50	50	static inline int END_XMIT(char *data);
…	…
80	80
81	81	/* Open IPCs */
82		writer_ipc =
83		sipc_open(SIPC_SHM_KEY, SIPC_SENDER, SIPC_SYSV_SHM, DATA_LEN);
	82	writer_ipc = sipc_open(SIPC_SHM_KEY, SIPC_SENDER, SIPC_SYSV_SHM, DATA_LEN);
84	83	if (writer_ipc == NULL) {
85	84	fprintf(stderr, "Could not initialize IPC resource\n");
86	85	return -1;
87	86	}
88		reader_ipc =
89		sipc_open(SIPC_SHM_KEY, SIPC_RECEIVER, SIPC_SYSV_SHM, DATA_LEN);
	87	reader_ipc = sipc_open(SIPC_SHM_KEY, SIPC_RECEIVER, SIPC_SYSV_SHM, DATA_LEN);
90	88	if (reader_ipc == NULL) {
91	89	fprintf(stderr, "Could not initialize IPC resource\n");
…	…
131	129	}
132	130
133		static void test_unforked_shm(void) {
	131	static void test_unforked_shm(void)
	132	{
134	133	static char message[] = "HELLO,";
135	134	size_t message_len = sizeof(message);
…	…
194	193
195	194	CU_TestInfo shm_tests[] = {
196		{"unforked shm", test_unforked_shm},
197		{"forked shm", test_shm},
198		{"binary shm", test_binary_shm},
	195	{"unforked shm", test_unforked_shm}
	196	,
	197	{"forked shm", test_shm}
	198	,
	199	{"binary shm", test_binary_shm}
	200	,
199	201	CU_TEST_INFO_NULL
200	202	};
…	…
213	215	}
214	216
215		if ((ifile = fopen(IN_FILE, "r")) == NULL) {
	217	if ((ifile = fopen(IN_FILE, "r")) == NULL) {
216	218	sipc_error(writer_ipc, "fopen: %s\n", strerror(errno));
217	219	return;
…	…
219	221	/* Read DATA_LEN bytes from file */
220	222	bzero(shm_data, DATA_LEN);
221		if ((rbytes = fread(shm_data, sizeof(char), DATA_LEN-1, ifile)) < 0) {
	223	if ((rbytes = fread(shm_data, sizeof(char), DATA_LEN - 1, ifile)) < 0) {
222	224	sipc_error(writer_ipc, "fread: %s\n", strerror(errno));
223	225	fclose(ifile);
…	…
322	324	}
323	325
324		static void send_end_xmit(sipc_t *sipc)
	326	static void send_end_xmit(sipc_t * sipc)
325	327	{
326	328	char *data = sipc_get_data_ptr(sipc);
…	…
352	354	}
353	355
354		rbytes = fread(data, sizeof(char), DATA_LEN-1, ifile);
	356	rbytes = fread(data, sizeof(char), DATA_LEN - 1, ifile);
355	357	if (rbytes < 0) {
356	358	fclose(ifile);

trunk/libsipc/tests/test_ipc.c

r34	r35
33	33	/* Add suites to registry */
34	34	CU_SuiteInfo suites[] = {
35		{"mq", test_mqueue_init, test_mqueue_cleanup, mqueue_tests},
36		{"shm", test_shm_init, test_shm_cleanup, shm_tests},
	35	{"mq", test_mqueue_init, test_mqueue_cleanup, mqueue_tests}
	36	,
	37	{"shm", test_shm_init, test_shm_cleanup, shm_tests}
	38	,
37	39	CU_SUITE_INFO_NULL
38	40	};
…	…
43	45	CU_basic_run_tests();
44	46
45		printf("# suites run: %d, # tests run: %d\n",
46		CU_get_number_of_suites_run(),
47		CU_get_number_of_tests_run());
	47	printf("# suites run: %d, # tests run: %d\n", CU_get_number_of_suites_run(), CU_get_number_of_tests_run());
48	48
49	49	CU_cleanup_registry();

Navigation

Legend:

trunk/libsipc/include/sipc/sipc.h

trunk/libsipc/src/mqueue_internal.c

trunk/libsipc/src/mqueue_internal.h

trunk/libsipc/src/shm_internal.c

trunk/libsipc/src/shm_internal.h

trunk/libsipc/src/sipc.c

trunk/libsipc/src/sipc_internal.h

trunk/libsipc/src/sipc_mqueue.c

trunk/libsipc/src/sipc_mqueue.h

trunk/libsipc/src/sipc_shm.c

trunk/libsipc/src/sipc_shm.h

trunk/libsipc/tests/ipc_creator.c

trunk/libsipc/tests/mqueue.c

trunk/libsipc/tests/shm.c

trunk/libsipc/tests/test_ipc.c

Download in other formats: