fileaio-posix.cpp@ 23364

Last change on this file since 23364 was 23364, checked in by vboxsync, 16 years ago
Runtime/Aio-posix: The error code from lio_listio is in errno
Property svn:eol-style set to `native` Property svn:keywords set to `Author Date Id Revision`
File size: 29.2 KB

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

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

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

Download in other formats: