![[BACK]](/cvsweb/icons/back.gif){kind=icon}
Return to msg_queue.c CVS log ![[TXT]](/cvsweb/icons/text.gif){kind=icon}
![[DIR]](/cvsweb/icons/dir.gif){kind=icon}
Up to [Synchronet] / sbbs / src / xpdev|
Return to msg_queue.c CVS log | Up to [Synchronet] / sbbs / src / xpdev |
1.1 root 1: /* msg_queue.c */
2:
3: /* Uni or Bi-directional FIFO message queue */
4:
5: /* $Id: msg_queue.c,v 1.11 2005/05/09 09:01:28 rswindell Exp $ */
6:
7: /****************************************************************************
8: * @format.tab-size 4 (Plain Text/Source Code File Header) *
9: * @format.use-tabs true (see http://www.synchro.net/ptsc_hdr.html) *
10: * *
11: * Copyright 2005 Rob Swindell - http://www.synchro.net/copyright.html *
12: * *
13: * This library is free software; you can redistribute it and/or *
14: * modify it under the terms of the GNU Lesser General Public License *
15: * as published by the Free Software Foundation; either version 2 *
16: * of the License, or (at your option) any later version. *
17: * See the GNU Lesser General Public License for more details: lgpl.txt or *
18: * http://www.fsf.org/copyleft/lesser.html *
19: * *
20: * Anonymous FTP access to the most recent released source is available at *
21: * ftp://vert.synchro.net, ftp://cvs.synchro.net and ftp://ftp.synchro.net *
22: * *
23: * Anonymous CVS access to the development source and modification history *
24: * is available at cvs.synchro.net:/cvsroot/sbbs, example: *
25: * cvs -d :pserver:[email protected]:/cvsroot/sbbs login *
26: * (just hit return, no password is necessary) *
27: * cvs -d :pserver:[email protected]:/cvsroot/sbbs checkout src *
28: * *
29: * For Synchronet coding style and modification guidelines, see *
30: * http://www.synchro.net/source.html *
31: * *
32: * You are encouraged to submit any modifications (preferably in Unix diff *
33: * format) via e-mail to [email protected] *
34: * *
35: * Note: If this box doesn't appear square, then you need to fix your tabs. *
36: ****************************************************************************/
37:
38: #include <stdlib.h> /* malloc */
39: #include <string.h> /* memset */
40:
41: #include "genwrap.h" /* msclock() */
42: #include "threadwrap.h" /* pthread_self */
43: #include "msg_queue.h"
44:
45: msg_queue_t* msgQueueInit(msg_queue_t* q, long flags)
46: {
47: if(q==NULL) {
48: if((q=(msg_queue_t*)malloc(sizeof(msg_queue_t)))==NULL)
49: return(NULL);
50: flags |= MSG_QUEUE_MALLOC;
51: }
52:
53: memset(q,0,sizeof(msg_queue_t));
54:
55: q->flags = flags;
56: q->refs = 1;
57: q->owner_thread_id = pthread_self();
58:
59: if(q->flags&MSG_QUEUE_BIDIR)
60: listInit(&q->in,LINK_LIST_SEMAPHORE);
61: listInit(&q->out,LINK_LIST_SEMAPHORE);
62:
63: return(q);
64: }
65:
66: BOOL msgQueueFree(msg_queue_t* q)
67: {
68: if(q==NULL)
69: return(FALSE);
70:
71: listFree(&q->in);
72: listFree(&q->out);
73:
74: if(q->flags&MSG_QUEUE_MALLOC)
75: free(q);
76:
77: return(TRUE);
78: }
79:
80: long msgQueueAttach(msg_queue_t* q)
81: {
82: if(q==NULL)
83: return(-1);
84:
85: q->refs++;
86:
87: return(q->refs);
88: }
89:
90: long msgQueueDetach(msg_queue_t* q)
91: {
92: int refs;
93:
94: if(q==NULL || q->refs<1)
95: return(-1);
96:
97: if((refs=--q->refs)==0)
98: msgQueueFree(q);
99:
100: return(refs);
101: }
102:
103: void* msgQueueSetPrivateData(msg_queue_t* q, void* p)
104: {
105: void* old;
106:
107: if(q==NULL)
108: return(NULL);
109:
110: old=q->private_data;
111: q->private_data=p;
112: return(old);
113: }
114:
115: void* msgQueueGetPrivateData(msg_queue_t* q)
116: {
117: if(q==NULL)
118: return(NULL);
119: return(q->private_data);
120: }
121:
122: static link_list_t* msgQueueReadList(msg_queue_t* q)
123: {
124: if(q==NULL)
125: return(NULL);
126:
127: if((q->flags&MSG_QUEUE_BIDIR)
128: && q->owner_thread_id == pthread_self())
129: return(&q->in);
130: return(&q->out);
131: }
132:
133: static link_list_t* msgQueueWriteList(msg_queue_t* q)
134: {
135: if(q==NULL)
136: return(NULL);
137:
138: if(!(q->flags&MSG_QUEUE_BIDIR)
139: || q->owner_thread_id == pthread_self())
140: return(&q->out);
141: return(&q->in);
142: }
143:
144: long msgQueueReadLevel(msg_queue_t* q)
145: {
146: return listCountNodes(msgQueueReadList(q));
147: }
148:
149: static BOOL list_wait(link_list_t* list, long timeout)
150: {
151: #if defined(LINK_LIST_THREADSAFE)
152: if(timeout<0) /* infinite */
153: return listSemWait(list);
154: if(timeout==0) /* poll */
155: return listSemTryWait(list);
156:
157: return listSemTryWaitBlock(list,timeout);
158: #else
159: clock_t start;
160: long count;
161:
162: start=msclock();
163: while((count=listCountNodes(list))==0) {
164: if(timeout==0)
165: break;
166: if(timeout>0 && msclock()-start > timeout)
167: break;
168: YIELD();
169: }
170: return(INT_TO_BOOL(count));
171: #endif
172: }
173:
174: BOOL msgQueueWait(msg_queue_t* q, long timeout)
175: {
176: BOOL result;
177: link_list_t* list = msgQueueReadList(q);
178:
179: if((result=list_wait(list,timeout))==TRUE)
180: #if defined(LINK_LIST_THREADSAFE)
181: listSemPost(list) /* Replace the semaphore we just cleared */
182: #endif
183: ;
184:
185: return(result);
186: }
187:
188: void* msgQueueRead(msg_queue_t* q, long timeout)
189: {
190: link_list_t* list = msgQueueReadList(q);
191:
192: list_wait(list,timeout);
193:
194: return listShiftNode(list);
195: }
196:
197: void* msgQueuePeek(msg_queue_t* q, long timeout)
198: {
199: link_list_t* list = msgQueueReadList(q);
200:
201: if(list_wait(list,timeout))
202: #if defined(LINK_LIST_THREADSAFE)
203: listSemPost(list) /* Replace the semaphore we just cleared */
204: #endif
205: ;
206:
207: return listNodeData(listFirstNode(list));
208: }
209:
210: void* msgQueueFind(msg_queue_t* q, const void* data, size_t length)
211: {
212: link_list_t* list = msgQueueReadList(q);
213: list_node_t* node;
214:
215: if((node=listFindNode(list,data,length))==NULL)
216: return(NULL);
217: return listRemoveNode(list,node,/* Free Data? */FALSE);
218: }
219:
220: list_node_t* msgQueueFirstNode(msg_queue_t* q)
221: {
222: return listFirstNode(msgQueueReadList(q));
223: }
224:
225: list_node_t* msgQueueLastNode(msg_queue_t* q)
226: {
227: return listLastNode(msgQueueReadList(q));
228: }
229:
230: long msgQueueWriteLevel(msg_queue_t* q)
231: {
232: return listCountNodes(msgQueueWriteList(q));
233: }
234:
235: BOOL msgQueueWrite(msg_queue_t* q, const void* data, size_t length)
236: {
237: return listPushNodeData(msgQueueWriteList(q),data,length)!=NULL;
238: }
239:
This archive runs on limited infrastructure. Preserving old code on modern bandwidth. Automated agents are requested to crawl responsibly.