fileaio-posix.cpp@ 23343

Last change on this file since 23343 was 23343, checked in by vboxsync, 16 years ago
AIO/Posix: Build fix
Property svn:eol-style set to `native` Property svn:keywords set to `Author Date Id Revision`
File size: 28.0 KB

Line
1	/* $Id: fileaio-posix.cpp 23343 2009-09-25 16:30:58Z 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	#if defined(RT_OS_DARWIN)
48	# include <sys/types.h>
49	# include <sys/sysctl.h> /* for sysctlbyname */
50	#endif
51	#include <aio.h>
52	#include <errno.h>
53	#include <time.h>
54
55	/*
56	* Linux does not define this value.
57	* Just define it with really big
58	* value.
59	*/
60	#ifndef AIO_LISTIO_MAX
61	# define AIO_LISTIO_MAX UINT32_MAX
62	#endif
63
64	/*******************************************************************************
65	* Structures and Typedefs *
66	*******************************************************************************/
67	/**
68	* Async I/O request state.
69	*/
70	typedef struct RTFILEAIOREQINTERNAL
71	{
72	/** The aio control block. FIRST ELEMENT! */
73	struct aiocb AioCB;
74	/** Next element in the chain. */
75	struct RTFILEAIOREQINTERNAL *pNext;
76	/** Previous element in the chain. */
77	struct RTFILEAIOREQINTERNAL *pPrev;
78	/** Current state the request is in. */
79	RTFILEAIOREQSTATE enmState;
80	/** Flag whether this is a flush request. */
81	bool fFlush;
82	/** Flag indicating if the request was canceled. */
83	volatile bool fCanceled;
84	/** Opaque user data. */
85	void *pvUser;
86	/** Number of bytes actually transfered. */
87	size_t cbTransfered;
88	/** Status code. */
89	int Rc;
90	/** Completion context we are assigned to. */
91	struct RTFILEAIOCTXINTERNAL *pCtxInt;
92	/** Entry in the waiting list the request is in. */
93	unsigned iWaitingList;
94	/** Magic value (RTFILEAIOREQ_MAGIC). */
95	uint32_t u32Magic;
96	} RTFILEAIOREQINTERNAL, *PRTFILEAIOREQINTERNAL;
97
98	/**
99	* Async I/O completion context state.
100	*/
101	typedef struct RTFILEAIOCTXINTERNAL
102	{
103	/** Current number of requests active on this context. */
104	volatile int32_t cRequests;
105	/** Maximum number of requests this context can handle. */
106	uint32_t cMaxRequests;
107	/** The ID of the thread which is currently waiting for requests. */
108	volatile RTTHREAD hThreadWait;
109	/** Flag whether the thread was woken up. */
110	volatile bool fWokenUp;
111	/** Flag whether the thread is currently waiting in the syscall. */
112	volatile bool fWaiting;
113	/** Magic value (RTFILEAIOCTX_MAGIC). */
114	uint32_t u32Magic;
115	/** Flag whether the thread was woken up due to a internal event. */
116	volatile bool fWokenUpInternal;
117	/** List of new requests which needs to be inserted into apReqs by the
118	* waiting thread. */
119	volatile PRTFILEAIOREQINTERNAL apReqsNewHead[5];
120	/** Special entry for requests which are canceled. Because only one
121	* request can be canceled at a time and the thread canceling the request
122	* has to wait we need only one entry. */
123	volatile PRTFILEAIOREQINTERNAL pReqToCancel;
124	/** Event semaphore the canceling thread is waiting for completion of
125	* the operation. */
126	RTSEMEVENT SemEventCancel;
127	/** Number of elements in the waiting list. */
128	unsigned cReqsWait;
129	/** First free slot in the waiting list. */
130	unsigned iFirstFree;
131	/** List of requests we are currently waiting on.
132	* Size depends on cMaxRequests. */
133	volatile PRTFILEAIOREQINTERNAL apReqs[1];
134	} RTFILEAIOCTXINTERNAL, *PRTFILEAIOCTXINTERNAL;
135
136	/**
137	* Internal worker for waking up the waiting thread.
138	*/
139	static void rtFileAioCtxWakeup(PRTFILEAIOCTXINTERNAL pCtxInt)
140	{
141	/*
142	* Read the thread handle before the status flag.
143	* If we read the handle after the flag we might
144	* end up with an invalid handle because the thread
145	* waiting in RTFileAioCtxWakeup() might get scheduled
146	* before we read the flag and returns.
147	* We can ensure that the handle is valid if fWaiting is true
148	* when reading the handle before the status flag.
149	*/
150	RTTHREAD hThread;
151	ASMAtomicReadHandle(&pCtxInt->hThreadWait, &hThread);
152	bool fWaiting = ASMAtomicReadBool(&pCtxInt->fWaiting);
153	if (fWaiting)
154	{
155	/*
156	* If a thread waits the handle must be valid.
157	* It is possible that the thread returns from
158	* aio_suspend() before the signal is send.
159	* This is no problem because we already set fWokenUp
160	* to true which will let the thread return VERR_INTERRUPTED
161	* and the next call to RTFileAioCtxWait() will not
162	* return VERR_INTERRUPTED because signals are not saved
163	* and will simply vanish if the destination thread can't
164	* receive it.
165	*/
166	Assert(hThread != NIL_RTTHREAD);
167	RTThreadPoke(hThread);
168	}
169	}
170
171	/**
172	* Internal worker processing events and inserting new requests into the waiting list.
173	*/
174	static int rtFileAioCtxProcessEvents(PRTFILEAIOCTXINTERNAL pCtxInt)
175	{
176	int rc = VINF_SUCCESS;
177
178	/* Process new requests first. */
179	bool fWokenUp = ASMAtomicXchgBool(&pCtxInt->fWokenUpInternal, false);
180	if (fWokenUp)
181	{
182	for (unsigned iSlot = 0; iSlot < RT_ELEMENTS(pCtxInt->apReqsNewHead); iSlot++)
183	{
184	PRTFILEAIOREQINTERNAL pReqHead = (PRTFILEAIOREQINTERNAL)ASMAtomicXchgPtr((void* volatile*)&pCtxInt->apReqsNewHead[iSlot],
185	NULL);
186
187	while (pReqHead)
188	{
189	pCtxInt->apReqs[pCtxInt->iFirstFree] = pReqHead;
190	pReqHead->iWaitingList = pCtxInt->iFirstFree;
191	pReqHead = pReqHead->pNext;
192
193	/* Clear pointer to next and previous element just for safety. */
194	pCtxInt->apReqs[pCtxInt->iFirstFree]->pNext = NULL;
195	pCtxInt->apReqs[pCtxInt->iFirstFree]->pPrev = NULL;
196	pCtxInt->iFirstFree++;
197	Assert(pCtxInt->iFirstFree <= pCtxInt->cMaxRequests);
198	}
199	}
200
201	/* Check if a request needs to be canceled. */
202	PRTFILEAIOREQINTERNAL pReqToCancel = (PRTFILEAIOREQINTERNAL)ASMAtomicReadPtr((void* volatile*)&pCtxInt->pReqToCancel);
203	if (pReqToCancel)
204	{
205	/* Put it out of the waiting list. */
206	pCtxInt->apReqs[pReqToCancel->iWaitingList] = pCtxInt->apReqs[--pCtxInt->iFirstFree];
207	pCtxInt->apReqs[pReqToCancel->iWaitingList]->iWaitingList = pReqToCancel->iWaitingList;
208	ASMAtomicDecS32(&pCtxInt->cRequests);
209	RTSemEventSignal(pCtxInt->SemEventCancel);
210	}
211	}
212	else
213	{
214	if (ASMAtomicXchgBool(&pCtxInt->fWokenUp, false))
215	rc = VERR_INTERRUPTED;
216	}
217
218	return rc;
219	}
220
221	RTR3DECL(int) RTFileAioGetLimits(PRTFILEAIOLIMITS pAioLimits)
222	{
223	int rcBSD = 0;
224	AssertPtrReturn(pAioLimits, VERR_INVALID_POINTER);
225
226	#if defined(RT_OS_DARWIN)
227	int cReqsOutstandingMax = 0;
228	size_t cbParameter = sizeof(int);
229
230	rcBSD = sysctlbyname("kern.aioprocmax", /* name */
231	&cReqsOutstandingMax, /* Where to store the old value. */
232	&cbParameter, /* Size of the memory pointed to. */
233	NULL, /* Where the new value is located. */
234	NULL); /* Where the size of the new value is stored. */
235	if (rcBSD == -1)
236	return RTErrConvertFromErrno(errno);
237
238	pAioLimits->cReqsOutstandingMax = cReqsOutstandingMax;
239	pAioLimits->cbBufferAlignment = 0;
240	#else
241	pAioLimits->cReqsOutstandingMax = RTFILEAIO_UNLIMITED_REQS;
242	pAioLimits->cbBufferAlignment = 0;
243	#endif
244
245	return VINF_SUCCESS;
246	}
247
248	RTR3DECL(int) RTFileAioReqCreate(PRTFILEAIOREQ phReq)
249	{
250	AssertPtrReturn(phReq, VERR_INVALID_POINTER);
251
252	PRTFILEAIOREQINTERNAL pReqInt = (PRTFILEAIOREQINTERNAL)RTMemAllocZ(sizeof(RTFILEAIOREQINTERNAL));
253	if (RT_UNLIKELY(!pReqInt))
254	return VERR_NO_MEMORY;
255
256	pReqInt->pCtxInt = NULL;
257	pReqInt->u32Magic = RTFILEAIOREQ_MAGIC;
258	RTFILEAIOREQ_SET_STATE(pReqInt, COMPLETED);
259
260	*phReq = (RTFILEAIOREQ)pReqInt;
261
262	return VINF_SUCCESS;
263	}
264
265
266	RTDECL(int) RTFileAioReqDestroy(RTFILEAIOREQ hReq)
267	{
268	/*
269	* Validate the handle and ignore nil.
270	*/
271	if (hReq == NIL_RTFILEAIOREQ)
272	return VINF_SUCCESS;
273	PRTFILEAIOREQINTERNAL pReqInt = hReq;
274	RTFILEAIOREQ_VALID_RETURN(pReqInt);
275	RTFILEAIOREQ_NOT_STATE_RETURN_RC(pReqInt, SUBMITTED, VERR_FILE_AIO_IN_PROGRESS);
276
277	/*
278	* Trash the magic and free it.
279	*/
280	ASMAtomicUoWriteU32(&pReqInt->u32Magic, ~RTFILEAIOREQ_MAGIC);
281	RTMemFree(pReqInt);
282	return VINF_SUCCESS;
283	}
284
285	/**
286	* Worker setting up the request.
287	*/
288	DECLINLINE(int) rtFileAioReqPrepareTransfer(RTFILEAIOREQ hReq, RTFILE hFile,
289	unsigned uTransferDirection,
290	RTFOFF off, void *pvBuf, size_t cbTransfer,
291	void *pvUser)
292	{
293	/*
294	* Validate the input.
295	*/
296	PRTFILEAIOREQINTERNAL pReqInt = hReq;
297	RTFILEAIOREQ_VALID_RETURN(pReqInt);
298	RTFILEAIOREQ_NOT_STATE_RETURN_RC(pReqInt, SUBMITTED, VERR_FILE_AIO_IN_PROGRESS);
299	Assert(hFile != NIL_RTFILE);
300	AssertPtr(pvBuf);
301	Assert(off >= 0);
302	Assert(cbTransfer > 0);
303
304	memset(&pReqInt->AioCB, 0, sizeof(struct aiocb));
305	pReqInt->AioCB.aio_lio_opcode = uTransferDirection;
306	pReqInt->AioCB.aio_fildes = (int)hFile;
307	pReqInt->AioCB.aio_offset = off;
308	pReqInt->AioCB.aio_nbytes = cbTransfer;
309	pReqInt->AioCB.aio_buf = pvBuf;
310	pReqInt->pvUser = pvUser;
311	pReqInt->pCtxInt = NULL;
312	pReqInt->Rc = VERR_FILE_AIO_IN_PROGRESS;
313	RTFILEAIOREQ_SET_STATE(pReqInt, PREPARED);
314
315	return VINF_SUCCESS;
316	}
317
318
319	RTDECL(int) RTFileAioReqPrepareRead(RTFILEAIOREQ hReq, RTFILE hFile, RTFOFF off,
320	void pvBuf, size_t cbRead, void pvUser)
321	{
322	return rtFileAioReqPrepareTransfer(hReq, hFile, LIO_READ,
323	off, pvBuf, cbRead, pvUser);
324	}
325
326
327	RTDECL(int) RTFileAioReqPrepareWrite(RTFILEAIOREQ hReq, RTFILE hFile, RTFOFF off,
328	void pvBuf, size_t cbWrite, void pvUser)
329	{
330	return rtFileAioReqPrepareTransfer(hReq, hFile, LIO_WRITE,
331	off, pvBuf, cbWrite, pvUser);
332	}
333
334
335	RTDECL(int) RTFileAioReqPrepareFlush(RTFILEAIOREQ hReq, RTFILE hFile, void *pvUser)
336	{
337	PRTFILEAIOREQINTERNAL pReqInt = (PRTFILEAIOREQINTERNAL)hReq;
338
339	RTFILEAIOREQ_VALID_RETURN(pReqInt);
340	RTFILEAIOREQ_NOT_STATE_RETURN_RC(pReqInt, SUBMITTED, VERR_FILE_AIO_IN_PROGRESS);
341	Assert(hFile != NIL_RTFILE);
342
343	pReqInt->fFlush = true;
344	pReqInt->AioCB.aio_fildes = (int)hFile;
345	pReqInt->pvUser = pvUser;
346	RTFILEAIOREQ_SET_STATE(pReqInt, PREPARED);
347
348	return VINF_SUCCESS;
349	}
350
351
352	RTDECL(void *) RTFileAioReqGetUser(RTFILEAIOREQ hReq)
353	{
354	PRTFILEAIOREQINTERNAL pReqInt = hReq;
355	RTFILEAIOREQ_VALID_RETURN_RC(pReqInt, NULL);
356
357	return pReqInt->pvUser;
358	}
359
360
361	RTDECL(int) RTFileAioReqCancel(RTFILEAIOREQ hReq)
362	{
363	PRTFILEAIOREQINTERNAL pReqInt = hReq;
364	RTFILEAIOREQ_VALID_RETURN(pReqInt);
365	RTFILEAIOREQ_STATE_RETURN_RC(pReqInt, SUBMITTED, VERR_FILE_AIO_NOT_SUBMITTED);
366
367	ASMAtomicXchgBool(&pReqInt->fCanceled, true);
368
369	int rcPosix = aio_cancel(pReqInt->AioCB.aio_fildes, &pReqInt->AioCB);
370
371	if (rcPosix == AIO_CANCELED)
372	{
373	PRTFILEAIOCTXINTERNAL pCtxInt = pReqInt->pCtxInt;
374	/*
375	* Notify the waiting thread that the request was canceled.
376	*/
377	AssertMsg(VALID_PTR(pCtxInt),
378	("Invalid state. Request was canceled but wasn't submitted\n"));
379
380	Assert(!pCtxInt->pReqToCancel);
381	ASMAtomicWritePtr((void* volatile*)&pCtxInt->pReqToCancel, pReqInt);
382	rtFileAioCtxWakeup(pCtxInt);
383
384	/* Wait for acknowledge. */
385	int rc = RTSemEventWait(pCtxInt->SemEventCancel, RT_INDEFINITE_WAIT);
386	AssertRC(rc);
387
388	ASMAtomicWritePtr((void* volatile*)&pCtxInt->pReqToCancel, NULL);
389	pReqInt->Rc = VERR_FILE_AIO_CANCELED;
390	RTFILEAIOREQ_SET_STATE(pReqInt, COMPLETED);
391	return VINF_SUCCESS;
392	}
393	else if (rcPosix == AIO_ALLDONE)
394	return VERR_FILE_AIO_COMPLETED;
395	else if (rcPosix == AIO_NOTCANCELED)
396	return VERR_FILE_AIO_IN_PROGRESS;
397	else
398	return RTErrConvertFromErrno(errno);
399	}
400
401
402	RTDECL(int) RTFileAioReqGetRC(RTFILEAIOREQ hReq, size_t *pcbTransfered)
403	{
404	PRTFILEAIOREQINTERNAL pReqInt = hReq;
405	RTFILEAIOREQ_VALID_RETURN(pReqInt);
406	RTFILEAIOREQ_NOT_STATE_RETURN_RC(pReqInt, SUBMITTED, VERR_FILE_AIO_IN_PROGRESS);
407	RTFILEAIOREQ_NOT_STATE_RETURN_RC(pReqInt, PREPARED, VERR_FILE_AIO_NOT_SUBMITTED);
408	AssertPtrNull(pcbTransfered);
409
410	if ( (RT_SUCCESS(pReqInt->Rc))
411	&& (pcbTransfered))
412	*pcbTransfered = pReqInt->cbTransfered;
413
414	return pReqInt->Rc;
415	}
416
417
418	RTDECL(int) RTFileAioCtxCreate(PRTFILEAIOCTX phAioCtx, uint32_t cAioReqsMax)
419	{
420	PRTFILEAIOCTXINTERNAL pCtxInt;
421	AssertPtrReturn(phAioCtx, VERR_INVALID_POINTER);
422
423	if (cAioReqsMax == RTFILEAIO_UNLIMITED_REQS)
424	return VERR_OUT_OF_RANGE;
425
426	pCtxInt = (PRTFILEAIOCTXINTERNAL)RTMemAllocZ( sizeof(RTFILEAIOCTXINTERNAL)
427	+ cAioReqsMax * sizeof(PRTFILEAIOREQINTERNAL));
428	if (RT_UNLIKELY(!pCtxInt))
429	return VERR_NO_MEMORY;
430
431	/* Create event semaphore. */
432	int rc = RTSemEventCreate(&pCtxInt->SemEventCancel);
433	if (RT_FAILURE(rc))
434	{
435	RTMemFree(pCtxInt);
436	return rc;
437	}
438
439	pCtxInt->u32Magic = RTFILEAIOCTX_MAGIC;
440	pCtxInt->cMaxRequests = cAioReqsMax;
441	*phAioCtx = (RTFILEAIOCTX)pCtxInt;
442
443	return VINF_SUCCESS;
444	}
445
446
447	RTDECL(int) RTFileAioCtxDestroy(RTFILEAIOCTX hAioCtx)
448	{
449	PRTFILEAIOCTXINTERNAL pCtxInt = hAioCtx;
450
451	AssertPtrReturn(pCtxInt, VERR_INVALID_HANDLE);
452
453	if (RT_UNLIKELY(pCtxInt->cRequests))
454	return VERR_FILE_AIO_BUSY;
455
456	RTSemEventDestroy(pCtxInt->SemEventCancel);
457	RTMemFree(pCtxInt);
458
459	return VINF_SUCCESS;
460	}
461
462
463	RTDECL(uint32_t) RTFileAioCtxGetMaxReqCount(RTFILEAIOCTX hAioCtx)
464	{
465	PRTFILEAIOCTXINTERNAL pCtxInt = hAioCtx;
466
467	if (hAioCtx == NIL_RTFILEAIOCTX)
468	return RTFILEAIO_UNLIMITED_REQS;
469	else
470	return pCtxInt->cMaxRequests;
471	}
472
473	RTDECL(int) RTFileAioCtxAssociateWithFile(RTFILEAIOCTX hAioCtx, RTFILE hFile)
474	{
475	return VINF_SUCCESS;
476	}
477
478	RTDECL(int) RTFileAioCtxSubmit(RTFILEAIOCTX hAioCtx, PRTFILEAIOREQ pahReqs, size_t cReqs)
479	{
480	int rc = VINF_SUCCESS;
481	PRTFILEAIOCTXINTERNAL pCtxInt = hAioCtx;
482
483	/* Parameter checks */
484	AssertPtrReturn(pCtxInt, VERR_INVALID_HANDLE);
485	AssertReturn(cReqs != 0, VERR_INVALID_POINTER);
486	AssertPtrReturn(pahReqs, VERR_INVALID_PARAMETER);
487
488	/* Check that we don't exceed the limit */
489	if (ASMAtomicUoReadS32(&pCtxInt->cRequests) + cReqs > pCtxInt->cMaxRequests)
490	return VERR_FILE_AIO_LIMIT_EXCEEDED;
491
492	PRTFILEAIOREQINTERNAL pHead = NULL;
493
494	do
495	{
496	int rcPosix = 0;
497	size_t cReqsSubmit = 0;
498	size_t i = 0;
499	PRTFILEAIOREQINTERNAL pReqInt;
500
501	while ( (i < cReqs)
502	&& (i < AIO_LISTIO_MAX))
503	{
504	pReqInt = pahReqs[i];
505	if (RTFILEAIOREQ_IS_NOT_VALID(pReqInt))
506	{
507	/* Undo everything and stop submitting. */
508	for (size_t iUndo = 0; iUndo < i; iUndo++)
509	{
510	pReqInt = pahReqs[iUndo];
511	RTFILEAIOREQ_SET_STATE(pReqInt, PREPARED);
512	pReqInt->pCtxInt = NULL;
513
514	/* Unlink from the list again. */
515	PRTFILEAIOREQINTERNAL pNext, pPrev;
516	pNext = pReqInt->pNext;
517	pPrev = pReqInt->pPrev;
518	if (pNext)
519	pNext->pPrev = pPrev;
520	if (pPrev)
521	pPrev->pNext = pNext;
522	else
523	pHead = pNext;
524	}
525	rc = VERR_INVALID_HANDLE;
526	break;
527	}
528
529	pReqInt->pCtxInt = pCtxInt;
530
531	/* Link them together. */
532	pReqInt->pNext = pHead;
533	if (pHead)
534	pHead->pPrev = pReqInt;
535	pReqInt->pPrev = NULL;
536	pHead = pReqInt;
537	RTFILEAIOREQ_SET_STATE(pReqInt, SUBMITTED);
538
539	if (pReqInt->fFlush)
540	break;
541
542	cReqsSubmit++;
543	i++;
544	}
545
546	if (cReqsSubmit)
547	{
548	rcPosix = lio_listio(LIO_NOWAIT, (struct aiocb **)pahReqs, cReqsSubmit, NULL);
549	if (RT_UNLIKELY(rcPosix < 0))
550	{
551	if (rcPosix == EAGAIN)
552	rc = VERR_FILE_AIO_INSUFFICIENT_RESSOURCES;
553	else
554	rc = RTErrConvertFromErrno(errno);
555
556	/* Check which ones were not submitted. */
557	for (i = 0; i < cReqs; i++)
558	{
559	pReqInt = pahReqs[i];
560
561	rcPosix = aio_error(&pReqInt->AioCB);
562	Assert(rcPosix != 0);
563
564	if (rcPosix != EINPROGRESS)
565	{
566	cReqsSubmit--;
567
568	if (rcPosix == EINVAL)
569	{
570	/* Was not submitted. */
571	RTFILEAIOREQ_SET_STATE(pReqInt, PREPARED);
572	}
573	else
574	{
575	/* An error occurred. */
576	RTFILEAIOREQ_SET_STATE(pReqInt, COMPLETED);
577
578	/*
579	* Looks like Apple and glibc interpret the standard in different ways.
580	* glibc returns the error code which would be in errno but Apple returns
581	* -1 and sets errno to the appropriate value
582	*/
583	#if defined(RT_OS_DARWIN)
584	Assert(rcPosix == -1);
585	pReqInt->Rc = RTErrConvertFromErrno(errno);
586	#elif defined(RT_OS_LINUX)
587	pReqInt->Rc = RTErrConvertFromErrno(rcPosix);
588	#endif
589	pReqInt->cbTransfered = 0;
590	}
591	/* Unlink from the list. */
592	PRTFILEAIOREQINTERNAL pNext, pPrev;
593	pNext = pReqInt->pNext;
594	pPrev = pReqInt->pPrev;
595	if (pNext)
596	pNext->pPrev = pPrev;
597	if (pPrev)
598	pPrev->pNext = pNext;
599	else
600	pHead = pNext;
601	}
602	}
603
604	break;
605	}
606
607	ASMAtomicAddS32(&pCtxInt->cRequests, cReqsSubmit);
608	cReqs -= cReqsSubmit;
609	pahReqs += cReqsSubmit;
610	}
611
612	/*
613	* Check if we have a flush request now.
614	* If not we hit the AIO_LISTIO_MAX limit
615	* and will continue submitting requests
616	* above.
617	*/
618	if (cReqs)
619	{
620	pReqInt = pahReqs[0];
621	RTFILEAIOREQ_VALID_RETURN(pReqInt);
622
623
624	if (pReqInt->fFlush)
625	{
626	/*
627	* lio_listio does not work with flush requests so
628	* we have to use aio_fsync directly.
629	*/
630	rcPosix = aio_fsync(O_SYNC, &pReqInt->AioCB);
631	if (RT_UNLIKELY(rcPosix < 0))
632	{
633	rc = RTErrConvertFromErrno(errno);
634	RTFILEAIOREQ_SET_STATE(pReqInt, COMPLETED);
635	pReqInt->Rc = rc;
636	pReqInt->cbTransfered = 0;
637
638	/* Unlink from the list. */
639	PRTFILEAIOREQINTERNAL pNext, pPrev;
640	pNext = pReqInt->pNext;
641	pPrev = pReqInt->pPrev;
642	if (pNext)
643	pNext->pPrev = pPrev;
644	if (pPrev)
645	pPrev->pNext = pNext;
646	else
647	pHead = pNext;
648	break;
649	}
650
651	ASMAtomicIncS32(&pCtxInt->cRequests);
652	cReqs--;
653	pahReqs++;
654	}
655	}
656	} while (cReqs);
657
658	if (pHead)
659	{
660	/*
661	* Forward successfully submitted requests to the thread waiting for requests.
662	* We search for a free slot first and if we don't find one
663	* we will grab the first one and append our list to the existing entries.
664	*/
665	unsigned iSlot = 0;
666	while ( (iSlot < RT_ELEMENTS(pCtxInt->apReqsNewHead))
667	&& !ASMAtomicCmpXchgPtr((void * volatile *)&pCtxInt->apReqsNewHead[iSlot], pHead, NULL))
668	iSlot++;
669
670	if (iSlot == RT_ELEMENTS(pCtxInt->apReqsNewHead))
671	{
672	/* Nothing found. */
673	PRTFILEAIOREQINTERNAL pOldHead = (PRTFILEAIOREQINTERNAL)ASMAtomicXchgPtr((void * volatile *)&pCtxInt->apReqsNewHead[0],
674	NULL);
675
676	/* Find the end of the current head and link the old list to the current. */
677	PRTFILEAIOREQINTERNAL pTail = pHead;
678	while (pTail->pNext)
679	pTail = pTail->pNext;
680
681	pTail->pNext = pOldHead;
682
683	ASMAtomicXchgPtr((void * volatile *)&pCtxInt->apReqsNewHead[0], pHead);
684	}
685
686	/* Set the internal wakeup flag and wakeup the thread if possible. */
687	bool fWokenUp = ASMAtomicXchgBool(&pCtxInt->fWokenUpInternal, true);
688	if (!fWokenUp)
689	rtFileAioCtxWakeup(pCtxInt);
690	}
691
692	return rc;
693	}
694
695
696	RTDECL(int) RTFileAioCtxWait(RTFILEAIOCTX hAioCtx, size_t cMinReqs, unsigned cMillisTimeout,
697	PRTFILEAIOREQ pahReqs, size_t cReqs, uint32_t *pcReqs)
698	{
699	int rc = VINF_SUCCESS;
700	int cRequestsCompleted = 0;
701	PRTFILEAIOCTXINTERNAL pCtxInt = (PRTFILEAIOCTXINTERNAL)hAioCtx;
702	struct timespec Timeout;
703	struct timespec *pTimeout = NULL;
704	uint64_t StartNanoTS = 0;
705
706	/* Check parameters. */
707	AssertPtrReturn(pCtxInt, VERR_INVALID_HANDLE);
708	AssertPtrReturn(pcReqs, VERR_INVALID_POINTER);
709	AssertPtrReturn(pahReqs, VERR_INVALID_POINTER);
710	AssertReturn(cReqs != 0, VERR_INVALID_PARAMETER);
711	AssertReturn(cReqs >= cMinReqs, VERR_OUT_OF_RANGE);
712
713	if (RT_UNLIKELY(ASMAtomicReadS32(&pCtxInt->cRequests) == 0))
714	return VERR_FILE_AIO_NO_REQUEST;
715
716	if (cMillisTimeout != RT_INDEFINITE_WAIT)
717	{
718	Timeout.tv_sec = cMillisTimeout / 1000;
719	Timeout.tv_nsec = (cMillisTimeout % 1000) * 1000000;
720	pTimeout = &Timeout;
721	StartNanoTS = RTTimeNanoTS();
722	}
723
724	/* Wait for at least one. */
725	if (!cMinReqs)
726	cMinReqs = 1;
727
728	/* For the wakeup call. */
729	Assert(pCtxInt->hThreadWait == NIL_RTTHREAD);
730	ASMAtomicWriteHandle(&pCtxInt->hThreadWait, RTThreadSelf());
731
732	/* Update the waiting list once before we enter the loop. */
733	rc = rtFileAioCtxProcessEvents(pCtxInt);
734
735	while ( cMinReqs
736	&& RT_SUCCESS_NP(rc))
737	{
738	ASMAtomicXchgBool(&pCtxInt->fWaiting, true);
739	int rcPosix = aio_suspend((const struct aiocb * const *)pCtxInt->apReqs,
740	pCtxInt->iFirstFree, pTimeout);
741	ASMAtomicXchgBool(&pCtxInt->fWaiting, false);
742	if (rcPosix < 0)
743	{
744	/* Check that this is an external wakeup event. */
745	if (errno == EINTR)
746	rc = rtFileAioCtxProcessEvents(pCtxInt);
747	else
748	rc = RTErrConvertFromErrno(errno);
749	}
750	else
751	{
752	/* Requests finished. */
753	unsigned iReqCurr = 0;
754	int cDone = 0;
755
756	/* Remove completed requests from the waiting list. */
757	while (iReqCurr < pCtxInt->iFirstFree)
758	{
759	PRTFILEAIOREQINTERNAL pReq = pCtxInt->apReqs[iReqCurr];
760	int rcReq = aio_error(&pReq->AioCB);
761
762	if (rcReq != EINPROGRESS)
763	{
764	/* Completed store the return code. */
765	if (rcReq == 0)
766	{
767	pReq->Rc = VINF_SUCCESS;
768	/* Call aio_return() to free ressources. */
769	pReq->cbTransfered = aio_return(&pReq->AioCB);
770	}
771	else
772	pReq->Rc = RTErrConvertFromErrno(rcReq);
773
774	/* Mark the request as finished. */
775	RTFILEAIOREQ_SET_STATE(pReq, COMPLETED);
776	cDone++;
777
778	/*
779	* Move the last entry into the current position to avoid holes
780	* but only if it is not the last element already.
781	*/
782	if (pReq->iWaitingList < pCtxInt->iFirstFree - 1)
783	{
784	pCtxInt->apReqs[pReq->iWaitingList] = pCtxInt->apReqs[--pCtxInt->iFirstFree];
785	pCtxInt->apReqs[pReq->iWaitingList]->iWaitingList = pReq->iWaitingList;
786	pCtxInt->apReqs[pCtxInt->iFirstFree] = NULL;
787	}
788	else
789	pCtxInt->iFirstFree--;
790
791	/* Put the request into the completed list. */
792	pahReqs[cRequestsCompleted++] = pReq;
793	}
794	else
795	iReqCurr++;
796	}
797
798	cReqs -= cDone;
799	cMinReqs -= cDone;
800	ASMAtomicSubS32(&pCtxInt->cRequests, cDone);
801
802	if ((cMillisTimeout != RT_INDEFINITE_WAIT) && (cMinReqs > 0))
803	{
804	uint64_t TimeDiff;
805
806	/* Recalculate the timeout. */
807	TimeDiff = RTTimeSystemNanoTS() - StartNanoTS;
808	Timeout.tv_sec = Timeout.tv_sec - (TimeDiff / 1000000);
809	Timeout.tv_nsec = Timeout.tv_nsec - (TimeDiff % 1000000);
810	}
811
812	/* Check for new elements. */
813	rc = rtFileAioCtxProcessEvents(pCtxInt);
814	}
815	}
816
817	*pcReqs = cRequestsCompleted;
818	Assert(pCtxInt->hThreadWait == RTThreadSelf());
819	ASMAtomicWriteHandle(&pCtxInt->hThreadWait, NIL_RTTHREAD);
820
821	return rc;
822	}
823
824
825	RTDECL(int) RTFileAioCtxWakeup(RTFILEAIOCTX hAioCtx)
826	{
827	PRTFILEAIOCTXINTERNAL pCtxInt = hAioCtx;
828	RTFILEAIOCTX_VALID_RETURN(pCtxInt);
829
830	/** @todo r=bird: Define the protocol for how to resume work after calling
831	* this function. */
832
833	bool fWokenUp = ASMAtomicXchgBool(&pCtxInt->fWokenUp, true);
834	if (!fWokenUp)
835	rtFileAioCtxWakeup(pCtxInt);
836
837	return VINF_SUCCESS;
838	}
839

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

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

Download in other formats: