fileaio-posix.cpp@ 59512

Last change on this file since 59512 was 57358, checked in by vboxsync, 9 years ago
*: scm cleanup run.
Property svn:eol-style set to `native` Property svn:keywords set to `Author Date Id Revision`
File size: 36.8 KB

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

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

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

Download in other formats: