fileaio-posix.cpp@ 19370

Last change on this file since 19370 was 19370, checked in by vboxsync, 16 years ago
Runtime/Aio: Fix the POSIX backend for Darwin. We have to obey the AIO_LISTIO_MAX limit when calling lio_listio. Submitting more requests than AIO_LISTIO_MAX makes the call fail
Property svn:eol-style set to `native` Property svn:keywords set to `Author Date Id Revision`
File size: 22.2 KB

Line
1	/* $Id: fileaio-posix.cpp 19370 2009-05-05 12:20:09Z vboxsync $ */
2	/** @file
3	* IPRT - File async I/O, native implementation for POSIX compliant host platforms.
4	*/
5
6	/*
7	* Copyright (C) 2006-2007 Sun Microsystems, Inc.
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	* Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa
27	* Clara, CA 95054 USA or visit http://www.sun.com if you need
28	* additional information or have any questions.
29	*/
30
31
32	/*******************************************************************************
33	* Header Files *
34	*******************************************************************************/
35	#define LOG_GROUP RTLOGGROUP_DIR
36	#include <iprt/asm.h>
37	#include <iprt/file.h>
38	#include <iprt/mem.h>
39	#include <iprt/assert.h>
40	#include <iprt/string.h>
41	#include <iprt/err.h>
42	#include <iprt/log.h>
43	#include <iprt/thread.h>
44	#include <iprt/semaphore.h>
45	#include "internal/fileaio.h"
46
47	#include <aio.h>
48	#include <errno.h>
49	#include <time.h>
50
51	/*
52	* Linux does not define this value.
53	* Just define it with really big
54	* value.
55	*/
56	#ifndef AIO_LISTIO_MAX
57	# define AIO_LISTIO_MAX UINT32_MAX
58	#endif
59
60	/*******************************************************************************
61	* Structures and Typedefs *
62	*******************************************************************************/
63	/**
64	* Async I/O request state.
65	*/
66	typedef struct RTFILEAIOREQINTERNAL
67	{
68	/** The aio control block. FIRST ELEMENT! */
69	struct aiocb AioCB;
70	/** Next element in the chain. */
71	struct RTFILEAIOREQINTERNAL *pNext;
72	/** Flag whether this is a flush request. */
73	bool fFlush;
74	/** Flag indicating if the request was canceled. */
75	volatile bool fCanceled;
76	/** Opaque user data. */
77	void *pvUser;
78	/** Number of bytes actually transfered. */
79	size_t cbTransfered;
80	/** Status code. */
81	int Rc;
82	/** Completion context we are assigned to. */
83	struct RTFILEAIOCTXINTERNAL *pCtxInt;
84	/** Entry in the waiting list the request is in. */
85	unsigned iWaitingList;
86	/** Magic value (RTFILEAIOREQ_MAGIC). */
87	uint32_t u32Magic;
88	} RTFILEAIOREQINTERNAL, *PRTFILEAIOREQINTERNAL;
89
90	/**
91	* Async I/O completion context state.
92	*/
93	typedef struct RTFILEAIOCTXINTERNAL
94	{
95	/** Current number of requests active on this context. */
96	volatile int32_t cRequests;
97	/** Maximum number of requests this context can handle. */
98	uint32_t cMaxRequests;
99	/** The ID of the thread which is currently waiting for requests. */
100	volatile RTTHREAD hThreadWait;
101	/** Flag whether the thread was woken up. */
102	volatile bool fWokenUp;
103	/** Flag whether the thread is currently waiting in the syscall. */
104	volatile bool fWaiting;
105	/** Magic value (RTFILEAIOCTX_MAGIC). */
106	uint32_t u32Magic;
107	/** Flag whether the thread was woken up due to a internal event. */
108	volatile bool fWokenUpInternal;
109	/** List of new requests which needs to be inserted into apReqs by the
110	* waiting thread. */
111	volatile PRTFILEAIOREQINTERNAL apReqsNewHead[5];
112	/** Special entry for requests which are canceled. Because only one
113	* request can be canceled at a time and the thread canceling the request
114	* has to wait we need only one entry. */
115	volatile PRTFILEAIOREQINTERNAL pReqToCancel;
116	/** Event semaphore the canceling thread is waiting for completion of
117	* the operation. */
118	RTSEMEVENT SemEventCancel;
119	/** Number of elements in the waiting list. */
120	unsigned cReqsWait;
121	/** First free slot in the waiting list. */
122	unsigned iFirstFree;
123	/** List of requests we are currently waiting on.
124	* Size depends on cMaxRequests. */
125	volatile PRTFILEAIOREQINTERNAL apReqs[1];
126	} RTFILEAIOCTXINTERNAL, *PRTFILEAIOCTXINTERNAL;
127
128	/**
129	* Internal worker for waking up the waiting thread.
130	*/
131	static void rtFileAioCtxWakeup(PRTFILEAIOCTXINTERNAL pCtxInt)
132	{
133	/*
134	* Read the thread handle before the status flag.
135	* If we read the handle after the flag we might
136	* end up with an invalid handle because the thread
137	* waiting in RTFileAioCtxWakeup() might get scheduled
138	* before we read the flag and returns.
139	* We can ensure that the handle is valid if fWaiting is true
140	* when reading the handle before the status flag.
141	*/
142	RTTHREAD hThread;
143	ASMAtomicReadHandle(&pCtxInt->hThreadWait, &hThread);
144	bool fWaiting = ASMAtomicReadBool(&pCtxInt->fWaiting);
145	if (fWaiting)
146	{
147	/*
148	* If a thread waits the handle must be valid.
149	* It is possible that the thread returns from
150	* aio_suspend() before the signal is send.
151	* This is no problem because we already set fWokenUp
152	* to true which will let the thread return VERR_INTERRUPTED
153	* and the next call to RTFileAioCtxWait() will not
154	* return VERR_INTERRUPTED because signals are not saved
155	* and will simply vanish if the destination thread can't
156	* receive it.
157	*/
158	Assert(hThread != NIL_RTTHREAD);
159	RTThreadPoke(hThread);
160	}
161	}
162
163	/**
164	* Internal worker processing events and inserting new requests into the waiting list.
165	*/
166	static int rtFileAioCtxProcessEvents(PRTFILEAIOCTXINTERNAL pCtxInt)
167	{
168	int rc = VINF_SUCCESS;
169
170	/* Process new requests first. */
171	bool fWokenUp = ASMAtomicXchgBool(&pCtxInt->fWokenUpInternal, false);
172	if (fWokenUp)
173	{
174	for (unsigned iSlot = 0; iSlot < RT_ELEMENTS(pCtxInt->apReqsNewHead); iSlot++)
175	{
176	PRTFILEAIOREQINTERNAL pReqHead = (PRTFILEAIOREQINTERNAL)ASMAtomicXchgPtr((void* volatile*)&pCtxInt->apReqsNewHead[iSlot],
177	NULL);
178
179	while (pReqHead)
180	{
181	pCtxInt->apReqs[pCtxInt->iFirstFree] = pReqHead;
182	pReqHead->iWaitingList = pCtxInt->iFirstFree;
183	pReqHead = pReqHead->pNext;
184
185	/* Clear pointer to next element just for safety. */
186	pCtxInt->apReqs[pCtxInt->iFirstFree]->pNext = NULL;
187	pCtxInt->iFirstFree++;
188	Assert(pCtxInt->iFirstFree <= pCtxInt->cMaxRequests);
189	}
190	}
191
192	/* Check if a request needs to be canceled. */
193	PRTFILEAIOREQINTERNAL pReqToCancel = (PRTFILEAIOREQINTERNAL)ASMAtomicReadPtr((void* volatile*)&pCtxInt->pReqToCancel);
194	if (pReqToCancel)
195	{
196	/* Put it out of the waiting list. */
197	pCtxInt->apReqs[pReqToCancel->iWaitingList] = pCtxInt->apReqs[--pCtxInt->iFirstFree];
198	pCtxInt->apReqs[pReqToCancel->iWaitingList]->iWaitingList = pReqToCancel->iWaitingList;
199	ASMAtomicDecS32(&pCtxInt->cRequests);
200	RTSemEventSignal(pCtxInt->SemEventCancel);
201	}
202	}
203	else
204	{
205	if (ASMAtomicXchgBool(&pCtxInt->fWokenUp, false))
206	rc = VERR_INTERRUPTED;
207	}
208
209	return rc;
210	}
211
212	RTR3DECL(int) RTFileAioReqCreate(PRTFILEAIOREQ phReq)
213	{
214	AssertPtrReturn(phReq, VERR_INVALID_POINTER);
215
216	PRTFILEAIOREQINTERNAL pReqInt = (PRTFILEAIOREQINTERNAL)RTMemAllocZ(sizeof(RTFILEAIOREQINTERNAL));
217	if (RT_UNLIKELY(!pReqInt))
218	return VERR_NO_MEMORY;
219
220	pReqInt->pCtxInt = NULL;
221	pReqInt->u32Magic = RTFILEAIOREQ_MAGIC;
222
223	*phReq = (RTFILEAIOREQ)pReqInt;
224
225	return VINF_SUCCESS;
226	}
227
228
229	RTDECL(void) RTFileAioReqDestroy(RTFILEAIOREQ hReq)
230	{
231	/*
232	* Validate the handle and ignore nil.
233	*/
234	if (hReq == NIL_RTFILEAIOREQ)
235	return;
236	PRTFILEAIOREQINTERNAL pReqInt = hReq;
237	RTFILEAIOREQ_VALID_RETURN_VOID(pReqInt);
238
239	/*
240	* Trash the magic and free it.
241	*/
242	ASMAtomicUoWriteU32(&pReqInt->u32Magic, ~RTFILEAIOREQ_MAGIC);
243	RTMemFree(pReqInt);
244	}
245
246	/**
247	* Worker setting up the request.
248	*/
249	DECLINLINE(int) rtFileAioReqPrepareTransfer(RTFILEAIOREQ hReq, RTFILE hFile,
250	unsigned uTransferDirection,
251	RTFOFF off, void *pvBuf, size_t cbTransfer,
252	void *pvUser)
253	{
254	/*
255	* Validate the input.
256	*/
257	PRTFILEAIOREQINTERNAL pReqInt = hReq;
258	RTFILEAIOREQ_VALID_RETURN(pReqInt);
259	Assert(hFile != NIL_RTFILE);
260	AssertPtr(pvBuf);
261	Assert(off >= 0);
262	Assert(cbTransfer > 0);
263
264	memset(&pReqInt->AioCB, 0, sizeof(struct aiocb));
265	pReqInt->AioCB.aio_lio_opcode = uTransferDirection;
266	pReqInt->AioCB.aio_fildes = (int)hFile;
267	pReqInt->AioCB.aio_offset = off;
268	pReqInt->AioCB.aio_nbytes = cbTransfer;
269	pReqInt->AioCB.aio_buf = pvBuf;
270	pReqInt->pvUser = pvUser;
271	pReqInt->pCtxInt = NULL;
272	pReqInt->Rc = VERR_FILE_AIO_IN_PROGRESS;
273
274	return VINF_SUCCESS;
275	}
276
277
278	RTDECL(int) RTFileAioReqPrepareRead(RTFILEAIOREQ hReq, RTFILE hFile, RTFOFF off,
279	void pvBuf, size_t cbRead, void pvUser)
280	{
281	return rtFileAioReqPrepareTransfer(hReq, hFile, LIO_READ,
282	off, pvBuf, cbRead, pvUser);
283	}
284
285
286	RTDECL(int) RTFileAioReqPrepareWrite(RTFILEAIOREQ hReq, RTFILE hFile, RTFOFF off,
287	void pvBuf, size_t cbWrite, void pvUser)
288	{
289	return rtFileAioReqPrepareTransfer(hReq, hFile, LIO_WRITE,
290	off, pvBuf, cbWrite, pvUser);
291	}
292
293
294	RTDECL(int) RTFileAioReqPrepareFlush(RTFILEAIOREQ hReq, RTFILE hFile, void *pvUser)
295	{
296	PRTFILEAIOREQINTERNAL pReqInt = (PRTFILEAIOREQINTERNAL)hReq;
297
298	RTFILEAIOREQ_VALID_RETURN(pReqInt);
299	Assert(hFile != NIL_RTFILE);
300
301	pReqInt->fFlush = true;
302	pReqInt->AioCB.aio_fildes = (int)hFile;
303	pReqInt->pvUser = pvUser;
304
305	return VINF_SUCCESS;
306	}
307
308
309	RTDECL(void *) RTFileAioReqGetUser(RTFILEAIOREQ hReq)
310	{
311	PRTFILEAIOREQINTERNAL pReqInt = hReq;
312	RTFILEAIOREQ_VALID_RETURN_RC(pReqInt, NULL);
313
314	return pReqInt->pvUser;
315	}
316
317
318	RTDECL(int) RTFileAioReqCancel(RTFILEAIOREQ hReq)
319	{
320	PRTFILEAIOREQINTERNAL pReqInt = hReq;
321	RTFILEAIOREQ_VALID_RETURN(pReqInt);
322
323	ASMAtomicXchgBool(&pReqInt->fCanceled, true);
324
325	int rcPosix = aio_cancel(pReqInt->AioCB.aio_fildes, &pReqInt->AioCB);
326
327	if (rcPosix == AIO_CANCELED)
328	{
329	PRTFILEAIOCTXINTERNAL pCtxInt = pReqInt->pCtxInt;
330	/*
331	* Notify the waiting thread that the request was canceled.
332	*/
333	AssertMsg(VALID_PTR(pCtxInt),
334	("Invalid state. Request was canceled but wasn't submitted\n"));
335
336	Assert(!pCtxInt->pReqToCancel);
337	ASMAtomicWritePtr((void* volatile*)&pCtxInt->pReqToCancel, pReqInt);
338	rtFileAioCtxWakeup(pCtxInt);
339
340	/* Wait for acknowledge. */
341	int rc = RTSemEventWait(pCtxInt->SemEventCancel, RT_INDEFINITE_WAIT);
342	AssertRC(rc);
343
344	ASMAtomicWritePtr((void* volatile*)&pCtxInt->pReqToCancel, NULL);
345	return VINF_SUCCESS;
346	}
347	else if (rcPosix == AIO_ALLDONE)
348	return VERR_FILE_AIO_COMPLETED;
349	else if (rcPosix == AIO_NOTCANCELED)
350	return VERR_FILE_AIO_IN_PROGRESS;
351	else
352	return RTErrConvertFromErrno(errno);
353	}
354
355
356	RTDECL(int) RTFileAioReqGetRC(RTFILEAIOREQ hReq, size_t *pcbTransfered)
357	{
358	PRTFILEAIOREQINTERNAL pReqInt = hReq;
359	RTFILEAIOREQ_VALID_RETURN(pReqInt);
360	AssertPtrNull(pcbTransfered);
361
362	if ( (pReqInt->Rc != VERR_FILE_AIO_IN_PROGRESS)
363	&& (pcbTransfered))
364	*pcbTransfered = pReqInt->cbTransfered;
365
366	return pReqInt->Rc;
367	}
368
369
370	RTDECL(int) RTFileAioCtxCreate(PRTFILEAIOCTX phAioCtx, uint32_t cAioReqsMax)
371	{
372	PRTFILEAIOCTXINTERNAL pCtxInt;
373	AssertPtrReturn(phAioCtx, VERR_INVALID_POINTER);
374
375	pCtxInt = (PRTFILEAIOCTXINTERNAL)RTMemAllocZ( sizeof(RTFILEAIOCTXINTERNAL)
376	+ cAioReqsMax * sizeof(PRTFILEAIOREQINTERNAL));
377	if (RT_UNLIKELY(!pCtxInt))
378	return VERR_NO_MEMORY;
379
380	/* Create event semaphore. */
381	int rc = RTSemEventCreate(&pCtxInt->SemEventCancel);
382	if (RT_FAILURE(rc))
383	{
384	RTMemFree(pCtxInt);
385	return rc;
386	}
387
388	pCtxInt->u32Magic = RTFILEAIOCTX_MAGIC;
389	pCtxInt->cMaxRequests = cAioReqsMax;
390	*phAioCtx = (RTFILEAIOCTX)pCtxInt;
391
392	return VINF_SUCCESS;
393	}
394
395
396	RTDECL(int) RTFileAioCtxDestroy(RTFILEAIOCTX hAioCtx)
397	{
398	PRTFILEAIOCTXINTERNAL pCtxInt = hAioCtx;
399
400	AssertPtrReturn(pCtxInt, VERR_INVALID_HANDLE);
401
402	if (RT_UNLIKELY(pCtxInt->cRequests))
403	return VERR_FILE_AIO_BUSY;
404
405	RTSemEventDestroy(pCtxInt->SemEventCancel);
406	RTMemFree(pCtxInt);
407
408	return VINF_SUCCESS;
409	}
410
411
412	RTDECL(uint32_t) RTFileAioCtxGetMaxReqCount(RTFILEAIOCTX hAioCtx)
413	{
414	PRTFILEAIOCTXINTERNAL pCtxInt = hAioCtx;
415
416	if (hAioCtx == NIL_RTFILEAIOCTX)
417	return RTFILEAIO_UNLIMITED_REQS;
418	else
419	return pCtxInt->cMaxRequests;
420	}
421
422	RTDECL(int) RTFileAioCtxAssociateWithFile(RTFILEAIOCTX hAioCtx, RTFILE hFile)
423	{
424	return VINF_SUCCESS;
425	}
426
427	RTDECL(int) RTFileAioCtxSubmit(RTFILEAIOCTX hAioCtx, PRTFILEAIOREQ pahReqs, size_t cReqs, size_t *pcReqs)
428	{
429	int rc = VINF_SUCCESS;
430	PRTFILEAIOCTXINTERNAL pCtxInt = hAioCtx;
431
432	/* Parameter checks */
433	AssertPtrReturn(pCtxInt, VERR_INVALID_HANDLE);
434	AssertReturn(cReqs != 0, VERR_INVALID_POINTER);
435	AssertPtrReturn(pahReqs, VERR_INVALID_PARAMETER);
436
437	/* Check that we don't exceed the limit */
438	if (ASMAtomicUoReadS32(&pCtxInt->cRequests) + cReqs > pCtxInt->cMaxRequests)
439	return VERR_FILE_AIO_LIMIT_EXCEEDED;
440
441	PRTFILEAIOREQINTERNAL pHead = NULL;
442
443	do
444	{
445	int rcPosix = 0;
446	size_t cReqsSubmit = 0;
447	size_t i = 0;
448	PRTFILEAIOREQINTERNAL pReqInt;
449
450	while ( (i < cReqs)
451	&& (i < AIO_LISTIO_MAX))
452	{
453	pReqInt = pahReqs[i];
454	RTFILEAIOREQ_VALID_RETURN(pReqInt);
455
456	pReqInt->pCtxInt = pCtxInt;
457
458	/* Link them together. */
459	pReqInt->pNext = pHead;
460	pHead = pReqInt;
461
462	if (pReqInt->fFlush)
463	break;
464
465	cReqsSubmit++;
466	i++;
467	}
468
469	if (cReqsSubmit)
470	{
471	rcPosix = lio_listio(LIO_NOWAIT, (struct aiocb **)pahReqs, cReqsSubmit, NULL);
472	if (RT_UNLIKELY(rcPosix < 0))
473	{
474	rc = RTErrConvertFromErrno(errno);
475	break;
476	}
477
478	ASMAtomicAddS32(&pCtxInt->cRequests, cReqsSubmit);
479	cReqs -= cReqsSubmit;
480	pahReqs += cReqsSubmit;
481	*pcReqs += cReqsSubmit;
482	}
483
484	/* Check if we have a flush request now. */
485	if (cReqs)
486	{
487	pReqInt = pahReqs[0];
488	RTFILEAIOREQ_VALID_RETURN(pReqInt);
489
490	if (pReqInt->fFlush)
491	{
492	/*
493	* lio_listio does not work with flush requests so
494	* we have to use aio_fsync directly.
495	*/
496	rcPosix = aio_fsync(O_SYNC, &pReqInt->AioCB);
497	if (RT_UNLIKELY(rcPosix < 0))
498	{
499	rc = RTErrConvertFromErrno(errno);
500	break;
501	}
502
503	ASMAtomicIncS32(&pCtxInt->cRequests);
504	cReqs--;
505	pahReqs++;
506	*pcReqs++;
507	}
508	}
509	} while (cReqs);
510
511	if (pHead)
512	{
513	/*
514	* Forward successfully submitted requests to the thread waiting for requests.
515	* We search for a free slot first and if we don't find one
516	* we will grab the first one and append our list to the existing entries.
517	*/
518	unsigned iSlot = 0;
519	while ( (iSlot < RT_ELEMENTS(pCtxInt->apReqsNewHead))
520	&& !ASMAtomicCmpXchgPtr((void * volatile *)&pCtxInt->apReqsNewHead[iSlot], pHead, NULL))
521	iSlot++;
522
523	if (iSlot == RT_ELEMENTS(pCtxInt->apReqsNewHead))
524	{
525	/* Nothing found. */
526	PRTFILEAIOREQINTERNAL pOldHead = (PRTFILEAIOREQINTERNAL)ASMAtomicXchgPtr((void * volatile *)&pCtxInt->apReqsNewHead[0],
527	NULL);
528
529	/* Find the end of the current head and link the old list to the current. */
530	PRTFILEAIOREQINTERNAL pTail = pHead;
531	while (pTail->pNext)
532	pTail = pTail->pNext;
533
534	pTail->pNext = pOldHead;
535
536	ASMAtomicXchgPtr((void * volatile *)&pCtxInt->apReqsNewHead[0], pHead);
537	}
538
539	/* Set the internal wakeup flag and wakeup the thread if possible. */
540	bool fWokenUp = ASMAtomicXchgBool(&pCtxInt->fWokenUpInternal, true);
541	if (!fWokenUp)
542	rtFileAioCtxWakeup(pCtxInt);
543	}
544
545	return rc;
546	}
547
548
549	RTDECL(int) RTFileAioCtxWait(RTFILEAIOCTX hAioCtx, size_t cMinReqs, unsigned cMillisTimeout,
550	PRTFILEAIOREQ pahReqs, size_t cReqs, uint32_t *pcReqs)
551	{
552	int rc = VINF_SUCCESS;
553	int cRequestsCompleted = 0;
554	PRTFILEAIOCTXINTERNAL pCtxInt = (PRTFILEAIOCTXINTERNAL)hAioCtx;
555	struct timespec Timeout;
556	struct timespec *pTimeout = NULL;
557	uint64_t StartNanoTS = 0;
558
559	/* Check parameters. */
560	AssertPtrReturn(pCtxInt, VERR_INVALID_HANDLE);
561	AssertPtrReturn(pcReqs, VERR_INVALID_POINTER);
562	AssertPtrReturn(pahReqs, VERR_INVALID_POINTER);
563	AssertReturn(cReqs != 0, VERR_INVALID_PARAMETER);
564	AssertReturn(cReqs >= cMinReqs, VERR_OUT_OF_RANGE);
565
566	if (RT_UNLIKELY(ASMAtomicReadS32(&pCtxInt->cRequests) == 0))
567	return VERR_FILE_AIO_NO_REQUEST;
568
569	if (cMillisTimeout != RT_INDEFINITE_WAIT)
570	{
571	Timeout.tv_sec = cMillisTimeout / 1000;
572	Timeout.tv_nsec = (cMillisTimeout % 1000) * 1000000;
573	pTimeout = &Timeout;
574	StartNanoTS = RTTimeNanoTS();
575	}
576
577	/* Wait for at least one. */
578	if (!cMinReqs)
579	cMinReqs = 1;
580
581	/* For the wakeup call. */
582	Assert(pCtxInt->hThreadWait == NIL_RTTHREAD);
583	ASMAtomicWriteHandle(&pCtxInt->hThreadWait, RTThreadSelf());
584
585	/* Update the waiting list once before we enter the loop. */
586	rc = rtFileAioCtxProcessEvents(pCtxInt);
587
588	while ( cMinReqs
589	&& RT_SUCCESS_NP(rc))
590	{
591	ASMAtomicXchgBool(&pCtxInt->fWaiting, true);
592	int rcPosix = aio_suspend((const struct aiocb * const *)pCtxInt->apReqs,
593	pCtxInt->iFirstFree, pTimeout);
594	ASMAtomicXchgBool(&pCtxInt->fWaiting, false);
595	if (rcPosix < 0)
596	{
597	/* Check that this is an external wakeup event. */
598	if (errno == EINTR)
599	rc = rtFileAioCtxProcessEvents(pCtxInt);
600	else
601	rc = RTErrConvertFromErrno(errno);
602	}
603	else
604	{
605	/* Requests finished. */
606	unsigned iReqCurr = 0;
607	int cDone = 0;
608
609	/* Remove completed requests from the waiting list. */
610	while (iReqCurr < pCtxInt->iFirstFree)
611	{
612	PRTFILEAIOREQINTERNAL pReq = pCtxInt->apReqs[iReqCurr];
613	int rcReq = aio_error(&pReq->AioCB);
614
615	if (rcReq != EINPROGRESS)
616	{
617	/* Completed store the return code. */
618	if (rcReq == 0)
619	{
620	pReq->Rc = VINF_SUCCESS;
621	/* Call aio_return() to free ressources. */
622	pReq->cbTransfered = aio_return(&pReq->AioCB);
623	}
624	else
625	pReq->Rc = RTErrConvertFromErrno(rcReq);
626
627	cDone++;
628
629	/*
630	* Move the last entry into the current position to avoid holes
631	* but only if it is not the last element already.
632	*/
633	if (pReq->iWaitingList < pCtxInt->iFirstFree - 1)
634	{
635	pCtxInt->apReqs[pReq->iWaitingList] = pCtxInt->apReqs[--pCtxInt->iFirstFree];
636	pCtxInt->apReqs[pReq->iWaitingList]->iWaitingList = pReq->iWaitingList;
637	pCtxInt->apReqs[pCtxInt->iFirstFree] = NULL;
638	}
639	else
640	pCtxInt->iFirstFree--;
641
642	/* Put the request into the completed list. */
643	pahReqs[cRequestsCompleted++] = pReq;
644	}
645	else
646	iReqCurr++;
647	}
648
649	cReqs -= cDone;
650	cMinReqs -= cDone;
651	ASMAtomicSubS32(&pCtxInt->cRequests, cDone);
652
653	if ((cMillisTimeout != RT_INDEFINITE_WAIT) && (cMinReqs > 0))
654	{
655	uint64_t TimeDiff;
656
657	/* Recalculate the timeout. */
658	TimeDiff = RTTimeSystemNanoTS() - StartNanoTS;
659	Timeout.tv_sec = Timeout.tv_sec - (TimeDiff / 1000000);
660	Timeout.tv_nsec = Timeout.tv_nsec - (TimeDiff % 1000000);
661	}
662
663	/* Check for new elements. */
664	rc = rtFileAioCtxProcessEvents(pCtxInt);
665	}
666	}
667
668	*pcReqs = cRequestsCompleted;
669	Assert(pCtxInt->hThreadWait == RTThreadSelf());
670	ASMAtomicWriteHandle(&pCtxInt->hThreadWait, NIL_RTTHREAD);
671
672	return rc;
673	}
674
675
676	RTDECL(int) RTFileAioCtxWakeup(RTFILEAIOCTX hAioCtx)
677	{
678	PRTFILEAIOCTXINTERNAL pCtxInt = hAioCtx;
679	RTFILEAIOCTX_VALID_RETURN(pCtxInt);
680
681	/** @todo r=bird: Define the protocol for how to resume work after calling
682	* this function. */
683
684	bool fWokenUp = ASMAtomicXchgBool(&pCtxInt->fWokenUp, true);
685	if (!fWokenUp)
686	rtFileAioCtxWakeup(pCtxInt);
687
688	return VINF_SUCCESS;
689	}
690

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

source: vbox/trunk/src/VBox/Runtime/r3/posix/fileaio-posix.cpp@ 19370

Download in other formats: