reqqueue.cpp@ 39500

Last change on this file since 39500 was 39500, checked in by vboxsync, 13 years ago
RTReq: More refactoring.
Property svn:eol-style set to `native` Property svn:keywords set to `Author Date Id Revision`
File size: 13.6 KB

Line
1	/* $Id: reqqueue.cpp 39500 2011-12-01 20:26:48Z vboxsync $ */
2	/** @file
3	* IPRT - Request packets
4	*/
5
6	/*
7	* Copyright (C) 2006-2011 Oracle Corporation
8	*
9	* This file is part of VirtualBox Open Source Edition (OSE), as
10	* available from http://www.215389.xyz. This file is free software;
11	* you can redistribute it and/or modify it under the terms of the GNU
12	* General Public License (GPL) as published by the Free Software
13	* Foundation, in version 2 as it comes in the "COPYING" file of the
14	* VirtualBox OSE distribution. VirtualBox OSE is distributed in the
15	* hope that it will be useful, but WITHOUT ANY WARRANTY of any kind.
16	*
17	* The contents of this file may alternatively be used under the terms
18	* of the Common Development and Distribution License Version 1.0
19	* (CDDL) only, as it comes in the "COPYING.CDDL" file of the
20	* VirtualBox OSE distribution, in which case the provisions of the
21	* CDDL are applicable instead of those of the GPL.
22	*
23	* You may elect to license modified versions of this file under the
24	* terms and conditions of either the GPL or the CDDL or both.
25	*/
26
27
28	/*******************************************************************************
29	* Header Files *
30	*******************************************************************************/
31	#include <iprt/req.h>
32	#include "internal/iprt.h"
33
34	#include <iprt/assert.h>
35	#include <iprt/asm.h>
36	#include <iprt/string.h>
37	#include <iprt/time.h>
38	#include <iprt/semaphore.h>
39	#include <iprt/thread.h>
40	#include <iprt/log.h>
41	#include <iprt/mem.h>
42
43	#include "internal/req.h"
44	#include "internal/magics.h"
45
46
47
48	RTDECL(int) RTReqQueueCreate(RTREQQUEUE *phQueue)
49	{
50	PRTREQQUEUEINT pQueue = (PRTREQQUEUEINT)RTMemAllocZ(sizeof(RTREQQUEUEINT));
51	if (!pQueue)
52	return VERR_NO_MEMORY;
53	int rc = RTSemEventCreate(&pQueue->EventSem);
54	if (RT_SUCCESS(rc))
55	{
56	pQueue->u32Magic = RTREQQUEUE_MAGIC;
57
58	*phQueue = pQueue;
59	return VINF_SUCCESS;
60	}
61
62	RTMemFree(pQueue);
63	return rc;
64	}
65	RT_EXPORT_SYMBOL(RTReqQueueCreate);
66
67
68	RTDECL(int) RTReqQueueDestroy(RTREQQUEUE hQueue)
69	{
70	/*
71	* Check input.
72	*/
73	if (hQueue == NIL_RTREQQUEUE)
74	return VINF_SUCCESS;
75	PRTREQQUEUEINT pQueue = hQueue;
76	AssertPtrReturn(pQueue, VERR_INVALID_HANDLE);
77	AssertReturn(ASMAtomicCmpXchgU32(&pQueue->u32Magic, RTREQQUEUE_MAGIC_DEAD, RTREQQUEUE_MAGIC), VERR_INVALID_HANDLE);
78
79	RTSemEventDestroy(pQueue->EventSem);
80	pQueue->EventSem = NIL_RTSEMEVENT;
81	RTMemFree(pQueue);
82	return VINF_SUCCESS;
83	}
84	RT_EXPORT_SYMBOL(RTReqQueueDestroy);
85
86
87	RTDECL(int) RTReqQueueProcess(RTREQQUEUE hQueue, RTMSINTERVAL cMillies)
88	{
89	LogFlow(("RTReqProcess %x\n", hQueue));
90
91	/*
92	* Check input.
93	*/
94	PRTREQQUEUEINT pQueue = hQueue;
95	AssertPtrReturn(pQueue, VERR_INVALID_HANDLE);
96	AssertReturn(pQueue->u32Magic == RTREQQUEUE_MAGIC, VERR_INVALID_HANDLE);
97
98	/*
99	* Process loop.
100	*
101	* We do not repeat the outer loop if we've got an informational status code
102	* since that code needs processing by our caller.
103	*/
104	int rc = VINF_SUCCESS;
105	while (rc <= VINF_SUCCESS)
106	{
107	/*
108	* Get pending requests.
109	*/
110	PRTREQ pReqs = ASMAtomicXchgPtrT(&pQueue->pReqs, NULL, PRTREQ);
111	if (!pReqs)
112	{
113	ASMAtomicWriteBool(&pQueue->fBusy, false); /* this aint 100% perfect, but it's good enough for now... */
114	/** @todo We currently don't care if the entire time wasted here is larger than
115	* cMillies */
116	rc = RTSemEventWait(pQueue->EventSem, cMillies);
117	if (rc != VINF_SUCCESS)
118	break;
119	continue;
120	}
121	ASMAtomicWriteBool(&pQueue->fBusy, true);
122
123	/*
124	* Reverse the list to process it in FIFO order.
125	*/
126	PRTREQ pReq = pReqs;
127	if (pReq->pNext)
128	Log2(("RTReqProcess: 2+ requests: %p %p %p\n", pReq, pReq->pNext, pReq->pNext->pNext));
129	pReqs = NULL;
130	while (pReq)
131	{
132	Assert(pReq->enmState == RTREQSTATE_QUEUED);
133	Assert(pReq->uOwner.hQueue == pQueue);
134	PRTREQ pCur = pReq;
135	pReq = pReq->pNext;
136	pCur->pNext = pReqs;
137	pReqs = pCur;
138	}
139
140
141	/*
142	* Process the requests.
143	*/
144	while (pReqs)
145	{
146	/* Unchain the first request and advance the list. */
147	pReq = pReqs;
148	pReqs = pReqs->pNext;
149	pReq->pNext = NULL;
150
151	/* Process the request */
152	rc = rtReqProcessOne(pReq);
153	AssertRC(rc);
154	if (rc != VINF_SUCCESS)
155	break; /** @todo r=bird: we're dropping requests here! Add 2nd queue that can hold them. (will fix when writing a testcase) */
156	}
157	}
158
159	LogFlow(("RTReqProcess: returns %Rrc\n", rc));
160	return rc;
161	}
162	RT_EXPORT_SYMBOL(RTReqQueueProcess);
163
164
165	RTDECL(int) RTReqQueueCall(RTREQQUEUE hQueue, PRTREQ *ppReq, RTMSINTERVAL cMillies, PFNRT pfnFunction, unsigned cArgs, ...)
166	{
167	va_list va;
168	va_start(va, cArgs);
169	int rc = RTReqQueueCallV(hQueue, ppReq, cMillies, RTREQFLAGS_IPRT_STATUS, pfnFunction, cArgs, va);
170	va_end(va);
171	return rc;
172	}
173	RT_EXPORT_SYMBOL(RTReqQueueCall);
174
175
176	RTDECL(int) RTReqQueueCallVoid(RTREQQUEUE hQueue, PRTREQ *ppReq, RTMSINTERVAL cMillies, PFNRT pfnFunction, unsigned cArgs, ...)
177	{
178	va_list va;
179	va_start(va, cArgs);
180	int rc = RTReqQueueCallV(hQueue, ppReq, cMillies, RTREQFLAGS_VOID, pfnFunction, cArgs, va);
181	va_end(va);
182	return rc;
183	}
184	RT_EXPORT_SYMBOL(RTReqQueueCallVoid);
185
186
187	RTDECL(int) RTReqQueueCallEx(RTREQQUEUE hQueue, PRTREQ *ppReq, RTMSINTERVAL cMillies, unsigned fFlags, PFNRT pfnFunction, unsigned cArgs, ...)
188	{
189	va_list va;
190	va_start(va, cArgs);
191	int rc = RTReqQueueCallV(hQueue, ppReq, cMillies, fFlags, pfnFunction, cArgs, va);
192	va_end(va);
193	return rc;
194	}
195	RT_EXPORT_SYMBOL(RTReqQueueCallEx);
196
197
198	RTDECL(int) RTReqQueueCallV(RTREQQUEUE hQueue, PRTREQ *ppReq, RTMSINTERVAL cMillies, unsigned fFlags, PFNRT pfnFunction, unsigned cArgs, va_list Args)
199	{
200	LogFlow(("RTReqCallV: cMillies=%d fFlags=%#x pfnFunction=%p cArgs=%d\n", cMillies, fFlags, pfnFunction, cArgs));
201
202	/*
203	* Check input.
204	*/
205	PRTREQQUEUEINT pQueue = hQueue;
206	AssertPtrReturn(pQueue, VERR_INVALID_HANDLE);
207	AssertReturn(pQueue->u32Magic == RTREQQUEUE_MAGIC, VERR_INVALID_HANDLE);
208	AssertPtrReturn(pfnFunction, VERR_INVALID_POINTER);
209	AssertReturn(!(fFlags & ~(RTREQFLAGS_RETURN_MASK \| RTREQFLAGS_NO_WAIT)), VERR_INVALID_PARAMETER);
210
211	if (!(fFlags & RTREQFLAGS_NO_WAIT) \|\| ppReq)
212	{
213	AssertPtrReturn(ppReq, VERR_INVALID_POINTER);
214	*ppReq = NULL;
215	}
216
217	PRTREQ pReq = NULL;
218	AssertMsgReturn(cArgs * sizeof(uintptr_t) <= sizeof(pReq->u.Internal.aArgs), ("cArgs=%u\n", cArgs), VERR_TOO_MUCH_DATA);
219
220	/*
221	* Allocate request
222	*/
223	int rc = RTReqQueueAlloc(pQueue, &pReq, RTREQTYPE_INTERNAL);
224	if (rc != VINF_SUCCESS)
225	return rc;
226
227	/*
228	* Initialize the request data.
229	*/
230	pReq->fFlags = fFlags;
231	pReq->u.Internal.pfn = pfnFunction;
232	pReq->u.Internal.cArgs = cArgs;
233	for (unsigned iArg = 0; iArg < cArgs; iArg++)
234	pReq->u.Internal.aArgs[iArg] = va_arg(Args, uintptr_t);
235
236	/*
237	* Queue the request and return.
238	*/
239	rc = RTReqSubmit(pReq, cMillies);
240	if ( rc != VINF_SUCCESS
241	&& rc != VERR_TIMEOUT)
242	{
243	RTReqFree(pReq);
244	pReq = NULL;
245	}
246	if (!(fFlags & RTREQFLAGS_NO_WAIT))
247	{
248	*ppReq = pReq;
249	LogFlow(("RTReqCallV: returns %Rrc *ppReq=%p\n", rc, pReq));
250	}
251	else
252	LogFlow(("RTReqCallV: returns %Rrc\n", rc));
253	Assert(rc != VERR_INTERRUPTED);
254	return rc;
255	}
256	RT_EXPORT_SYMBOL(RTReqQueueCallV);
257
258
259	RTDECL(bool) RTReqQueueIsBusy(RTREQQUEUE hQueue)
260	{
261	PRTREQQUEUEINT pQueue = hQueue;
262	AssertPtrReturn(pQueue, false);
263
264	if (ASMAtomicReadBool(&pQueue->fBusy))
265	return true;
266	if (ASMAtomicReadPtrT(&pQueue->pReqs, PRTREQ) != NULL)
267	return true;
268	if (ASMAtomicReadBool(&pQueue->fBusy))
269	return true;
270	return false;
271	}
272	RT_EXPORT_SYMBOL(RTReqQueueIsBusy);
273
274
275	/**
276	* Joins the list pList with whatever is linked up at *pHead.
277	*/
278	static void vmr3ReqJoinFreeSub(volatile PRTREQ *ppHead, PRTREQ pList)
279	{
280	for (unsigned cIterations = 0;; cIterations++)
281	{
282	PRTREQ pHead = ASMAtomicXchgPtrT(ppHead, pList, PRTREQ);
283	if (!pHead)
284	return;
285	PRTREQ pTail = pHead;
286	while (pTail->pNext)
287	pTail = pTail->pNext;
288	pTail->pNext = pList;
289	if (ASMAtomicCmpXchgPtr(ppHead, pHead, pList))
290	return;
291	pTail->pNext = NULL;
292	if (ASMAtomicCmpXchgPtr(ppHead, pHead, NULL))
293	return;
294	pList = pHead;
295	Assert(cIterations != 32);
296	Assert(cIterations != 64);
297	}
298	}
299
300
301	/**
302	* Joins the list pList with whatever is linked up at *pHead.
303	*/
304	static void vmr3ReqJoinFree(PRTREQQUEUEINT pQueue, PRTREQ pList)
305	{
306	/*
307	* Split the list if it's too long.
308	*/
309	unsigned cReqs = 1;
310	PRTREQ pTail = pList;
311	while (pTail->pNext)
312	{
313	if (cReqs++ > 25)
314	{
315	const uint32_t i = pQueue->iReqFree;
316	vmr3ReqJoinFreeSub(&pQueue->apReqFree[(i + 2) % RT_ELEMENTS(pQueue->apReqFree)], pTail->pNext);
317
318	pTail->pNext = NULL;
319	vmr3ReqJoinFreeSub(&pQueue->apReqFree[(i + 2 + (i == pQueue->iReqFree)) % RT_ELEMENTS(pQueue->apReqFree)], pTail->pNext);
320	return;
321	}
322	pTail = pTail->pNext;
323	}
324	vmr3ReqJoinFreeSub(&pQueue->apReqFree[(pQueue->iReqFree + 2) % RT_ELEMENTS(pQueue->apReqFree)], pList);
325	}
326
327
328	RTDECL(int) RTReqQueueAlloc(RTREQQUEUE hQueue, PRTREQ *ppReq, RTREQTYPE enmType)
329	{
330	/*
331	* Validate input.
332	*/
333	PRTREQQUEUEINT pQueue = hQueue;
334	AssertPtrReturn(pQueue, VERR_INVALID_HANDLE);
335	AssertReturn(pQueue->u32Magic == RTREQQUEUE_MAGIC, VERR_INVALID_HANDLE);
336	AssertMsgReturn(enmType > RTREQTYPE_INVALID && enmType < RTREQTYPE_MAX, ("%d\n", enmType), VERR_RT_REQUEST_INVALID_TYPE);
337
338	/*
339	* Try get a recycled packet.
340	* While this could all be solved with a single list with a lock, it's a sport
341	* of mine to avoid locks.
342	*/
343	int cTries = RT_ELEMENTS(pQueue->apReqFree) * 2;
344	while (--cTries >= 0)
345	{
346	PRTREQ volatile *ppHead = &pQueue->apReqFree[ASMAtomicIncU32(&pQueue->iReqFree) % RT_ELEMENTS(pQueue->apReqFree)];
347	#if 0 /* sad, but this won't work safely because the reading of pReq->pNext. */
348	PRTREQ pNext = NULL;
349	PRTREQ pReq = *ppHead;
350	if ( pReq
351	&& !ASMAtomicCmpXchgPtr(ppHead, (pNext = pReq->pNext), pReq)
352	&& (pReq = *ppHead)
353	&& !ASMAtomicCmpXchgPtr(ppHead, (pNext = pReq->pNext), pReq))
354	pReq = NULL;
355	if (pReq)
356	{
357	Assert(pReq->pNext == pNext); NOREF(pReq);
358	#else
359	PRTREQ pReq = ASMAtomicXchgPtrT(ppHead, NULL, PRTREQ);
360	if (pReq)
361	{
362	PRTREQ pNext = pReq->pNext;
363	if ( pNext
364	&& !ASMAtomicCmpXchgPtr(ppHead, pNext, NULL))
365	{
366	vmr3ReqJoinFree(pQueue, pReq->pNext);
367	}
368	#endif
369	ASMAtomicDecU32(&pQueue->cReqFree);
370
371	/*
372	* Make sure the event sem is not signaled.
373	*/
374	if (!pReq->fEventSemClear)
375	{
376	int rc = RTSemEventWait(pReq->EventSem, 0);
377	if (rc != VINF_SUCCESS && rc != VERR_TIMEOUT)
378	{
379	/*
380	* This shall not happen, but if it does we'll just destroy
381	* the semaphore and create a new one.
382	*/
383	AssertMsgFailed(("rc=%Rrc from RTSemEventWait(%#x).\n", rc, pReq->EventSem));
384	RTSemEventDestroy(pReq->EventSem);
385	rc = RTSemEventCreate(&pReq->EventSem);
386	AssertRC(rc);
387	if (rc != VINF_SUCCESS)
388	return rc;
389	}
390	pReq->fEventSemClear = true;
391	}
392	else
393	Assert(RTSemEventWait(pReq->EventSem, 0) == VERR_TIMEOUT);
394
395	/*
396	* Initialize the packet and return it.
397	*/
398	Assert(pReq->u32Magic == RTREQ_MAGIC);
399	Assert(pReq->enmType == RTREQTYPE_INVALID);
400	Assert(pReq->enmState == RTREQSTATE_FREE);
401	Assert(!pReq->fPoolOrQueue);
402	Assert(pReq->uOwner.hQueue == pQueue);
403	ASMAtomicWriteNullPtr(&pReq->pNext);
404	pReq->iStatus = VERR_RT_REQUEST_STATUS_STILL_PENDING;
405	pReq->enmState = RTREQSTATE_ALLOCATED;
406	pReq->fFlags = RTREQFLAGS_IPRT_STATUS;
407	pReq->enmType = enmType;
408
409	*ppReq = pReq;
410	LogFlow(("RTReqAlloc: returns VINF_SUCCESS *ppReq=%p recycled\n", pReq));
411	return VINF_SUCCESS;
412	}
413	}
414
415	/*
416	* Ok allocate one.
417	*/
418	PRTREQ pReq = (PRTREQ)RTMemAllocZ(sizeof(*pReq));
419	if (!pReq)
420	return VERR_NO_MEMORY;
421
422	/*
423	* Create the semaphore.
424	*/
425	int rc = RTSemEventCreate(&pReq->EventSem);
426	AssertRC(rc);
427	if (rc != VINF_SUCCESS)
428	{
429	RTMemFree(pReq);
430	return rc;
431	}
432
433	/*
434	* Initialize the packet and return it.
435	*/
436	pReq->u32Magic = RTREQ_MAGIC;
437	pReq->fEventSemClear= true;
438	pReq->fPoolOrQueue = false;
439	pReq->iStatus = VERR_RT_REQUEST_STATUS_STILL_PENDING;
440	pReq->enmState = RTREQSTATE_ALLOCATED;
441	pReq->pNext = NULL;
442	pReq->uOwner.hQueue = pQueue;
443	pReq->fFlags = RTREQFLAGS_IPRT_STATUS;
444	pReq->enmType = enmType;
445
446	*ppReq = pReq;
447	LogFlow(("RTReqAlloc: returns VINF_SUCCESS *ppReq=%p new\n", pReq));
448	return VINF_SUCCESS;
449	}
450	RT_EXPORT_SYMBOL(RTReqQueueAlloc);
451

Note: See TracBrowser for help on using the repository browser.

source: vbox/trunk/src/VBox/Runtime/common/misc/reqqueue.cpp@ 39500

Download in other formats: