aiomgr.cpp@ 49329

Last change on this file since 49329 was 46246, checked in by vboxsync, 12 years ago
Runtime/AioMgr: Bug fixes
Property svn:eol-style set to `native` Property svn:keywords set to `Author Date Id Revision`
File size: 41.4 KB

Line
1	/* $Id: aiomgr.cpp 46246 2013-05-23 19:15:44Z vboxsync $ */
2	/** @file
3	* IPRT - Async I/O manager.
4	*/
5
6	/*
7	* Copyright (C) 2013 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	* Header Files *
29	*******************************************************************************/
30
31	#include <iprt/aiomgr.h>
32	#include <iprt/err.h>
33	#include <iprt/asm.h>
34	#include <iprt/mem.h>
35	#include <iprt/file.h>
36	#include <iprt/list.h>
37	#include <iprt/thread.h>
38	#include <iprt/assert.h>
39	#include <iprt/string.h>
40	#include <iprt/critsect.h>
41	#include <iprt/memcache.h>
42	#include <iprt/semaphore.h>
43	#include <iprt/queueatomic.h>
44
45	#include "internal/magics.h"
46
47	/*******************************************************************************
48	* Structures and Typedefs *
49	*******************************************************************************/
50
51	/** Pointer to an internal async I/O file instance. */
52	typedef struct RTAIOMGRFILEINT *PRTAIOMGRFILEINT;
53
54	/**
55	* Blocking event types.
56	*/
57	typedef enum RTAIOMGREVENT
58	{
59	/** Invalid tye */
60	RTAIOMGREVENT_INVALID = 0,
61	/** No event pending. */
62	RTAIOMGREVENT_NO_EVENT,
63	/** A file is added to the manager. */
64	RTAIOMGREVENT_FILE_ADD,
65	/** A file is about to be closed. */
66	RTAIOMGREVENT_FILE_CLOSE,
67	/** The async I/O manager is shut down. */
68	RTAIOMGREVENT_SHUTDOWN,
69	/** 32bit hack */
70	RTAIOMGREVENT_32BIT_HACK = 0x7fffffff
71	} RTAIOMGREVENT;
72
73	/**
74	* Async I/O manager instance data.
75	*/
76	typedef struct RTAIOMGRINT
77	{
78	/** Magic value. */
79	uint32_t u32Magic;
80	/** Reference count. */
81	volatile uint32_t cRefs;
82	/** Async I/O context handle. */
83	RTFILEAIOCTX hAioCtx;
84	/** async I/O thread. */
85	RTTHREAD hThread;
86	/** List of files assigned to this manager. */
87	RTLISTANCHOR ListFiles;
88	/** Number of requests active currently. */
89	unsigned cReqsActive;
90	/** Number of maximum requests active. */
91	uint32_t cReqsActiveMax;
92	/** Memory cache for requests. */
93	RTMEMCACHE hMemCacheReqs;
94	/** Critical section protecting the blocking event handling. */
95	RTCRITSECT CritSectBlockingEvent;
96	/** Event semaphore for blocking external events.
97	* The caller waits on it until the async I/O manager
98	* finished processing the event. */
99	RTSEMEVENT hEventSemBlock;
100	/** Blocking event type */
101	volatile RTAIOMGREVENT enmBlockingEvent;
102	/** Event type data */
103	union
104	{
105	/** The file to be added */
106	volatile PRTAIOMGRFILEINT pFileAdd;
107	/** The file to be closed */
108	volatile PRTAIOMGRFILEINT pFileClose;
109	} BlockingEventData;
110	} RTAIOMGRINT;
111	/** Pointer to an internal async I/O manager instance. */
112	typedef RTAIOMGRINT *PRTAIOMGRINT;
113
114	/**
115	* Async I/O manager file instance data.
116	*/
117	typedef struct RTAIOMGRFILEINT
118	{
119	/** Magic value. */
120	uint32_t u32Magic;
121	/** Reference count. */
122	volatile uint32_t cRefs;
123	/** Flags. */
124	uint32_t fFlags;
125	/** Opaque user data passed on creation. */
126	void *pvUser;
127	/** File handle. */
128	RTFILE hFile;
129	/** async I/O manager this file belongs to. */
130	PRTAIOMGRINT pAioMgr;
131	/** Work queue for new requests. */
132	RTQUEUEATOMIC QueueReqs;
133	/** Completion callback for this file. */
134	PFNRTAIOMGRREQCOMPLETE pfnReqCompleted;
135	/** Data for exclusive use by the assigned async I/O manager. */
136	struct
137	{
138	/** List node of assigned files for a async I/O manager. */
139	RTLISTNODE NodeAioMgrFiles;
140	/** List of requests waiting for submission. */
141	RTLISTANCHOR ListWaitingReqs;
142	/** Number of requests currently being processed for this endpoint
143	* (excluded flush requests). */
144	unsigned cReqsActive;
145	} AioMgr;
146	} RTAIOMGRFILEINT;
147
148	/** Flag whether the file is closed. */
149	#define RTAIOMGRFILE_FLAGS_CLOSING RT_BIT_32(1)
150
151	/**
152	* Request type.
153	*/
154	typedef enum RTAIOMGRREQTYPE
155	{
156	/** Invalid request type. */
157	RTAIOMGRREQTYPE_INVALID = 0,
158	/** Read reques type. */
159	RTAIOMGRREQTYPE_READ,
160	/** Write request. */
161	RTAIOMGRREQTYPE_WRITE,
162	/** Flush request. */
163	RTAIOMGRREQTYPE_FLUSH,
164	/** Prefetech request. */
165	RTAIOMGRREQTYPE_PREFETCH,
166	/** 32bit hack. */
167	RTAIOMGRREQTYPE_32BIT_HACK = 0x7fffffff
168	} RTAIOMGRREQTYPE;
169	/** Pointer to a reques type. */
170	typedef RTAIOMGRREQTYPE *PRTAIOMGRREQTYPE;
171
172	/**
173	* Async I/O manager request.
174	*/
175	typedef struct RTAIOMGRREQ
176	{
177	/** Atomic queue work item. */
178	RTQUEUEATOMICITEM WorkItem;
179	/** Node for a waiting list. */
180	RTLISTNODE NodeWaitingList;
181	/** Request flags. */
182	uint32_t fFlags;
183	/** Transfer type. */
184	RTAIOMGRREQTYPE enmType;
185	/** Assigned file request. */
186	RTFILEAIOREQ hReqIo;
187	/** File the request belongs to. */
188	PRTAIOMGRFILEINT pFile;
189	/** Opaque user data. */
190	void *pvUser;
191	/** Start offset */
192	RTFOFF off;
193	/** Data segment. */
194	RTSGSEG DataSeg;
195	/** When non-zero the segment uses a bounce buffer because the provided buffer
196	* doesn't meet host requirements. */
197	size_t cbBounceBuffer;
198	/** Pointer to the used bounce buffer if any. */
199	void *pvBounceBuffer;
200	/** Start offset in the bounce buffer to copy from. */
201	uint32_t offBounceBuffer;
202	} RTAIOMGRREQ;
203	/** Pointer to a I/O manager request. */
204	typedef RTAIOMGRREQ *PRTAIOMGRREQ;
205
206	/** Flag whether the request was prepared already. */
207	#define RTAIOMGRREQ_FLAGS_PREPARED RT_BIT_32(0)
208
209	/*******************************************************************************
210	* Defined Constants And Macros *
211	*******************************************************************************/
212
213	/** Validates a handle and returns VERR_INVALID_HANDLE if not valid. */
214	#define RTAIOMGR_VALID_RETURN_RC(a_hAioMgr, a_rc) \
215	do { \
216	AssertPtrReturn((a_hAioMgr), (a_rc)); \
217	AssertReturn((a_hAioMgr)->u32Magic == RTAIOMGR_MAGIC, (a_rc)); \
218	} while (0)
219
220	/** Validates a handle and returns VERR_INVALID_HANDLE if not valid. */
221	#define RTAIOMGR_VALID_RETURN(a_hAioMgr) RTAIOMGR_VALID_RETURN_RC((hAioMgr), VERR_INVALID_HANDLE)
222
223	/** Validates a handle and returns (void) if not valid. */
224	#define RTAIOMGR_VALID_RETURN_VOID(a_hAioMgr) \
225	do { \
226	AssertPtrReturnVoid(a_hAioMgr); \
227	AssertReturnVoid((a_hAioMgr)->u32Magic == RTAIOMGR_MAGIC); \
228	} while (0)
229
230	/*******************************************************************************
231	* Internal Functions *
232	*******************************************************************************/
233
234	static int rtAioMgrReqsEnqueue(PRTAIOMGRINT pThis, PRTAIOMGRFILEINT pFile,
235	PRTFILEAIOREQ pahReqs, unsigned cReqs);
236
237	/**
238	* Removes an endpoint from the currently assigned manager.
239	*
240	* @returns TRUE if there are still requests pending on the current manager for this endpoint.
241	* FALSE otherwise.
242	* @param pEndpointRemove The endpoint to remove.
243	*/
244	static bool rtAioMgrFileRemove(PRTAIOMGRFILEINT pFile)
245	{
246	/* Make sure that there is no request pending on this manager for the endpoint. */
247	if (!pFile->AioMgr.cReqsActive)
248	{
249	RTListNodeRemove(&pFile->AioMgr.NodeAioMgrFiles);
250	return false;
251	}
252
253	return true;
254	}
255
256	/**
257	* Allocate a new I/O request.
258	*
259	* @returns Pointer to the allocated request or NULL if out of memory.
260	* @param pThis The async I/O manager instance.
261	*/
262	static PRTAIOMGRREQ rtAioMgrReqAlloc(PRTAIOMGRINT pThis)
263	{
264	return (PRTAIOMGRREQ)RTMemCacheAlloc(pThis->hMemCacheReqs);
265	}
266
267	/**
268	* Frees an I/O request.
269	*
270	* @returns nothing.
271	* @param pThis The async I/O manager instance.
272	* @param pReq The request to free.
273	*/
274	static void rtAioMgrReqFree(PRTAIOMGRINT pThis, PRTAIOMGRREQ pReq)
275	{
276	if (pReq->cbBounceBuffer)
277	{
278	AssertPtr(pReq->pvBounceBuffer);
279	RTMemPageFree(pReq->pvBounceBuffer, pReq->cbBounceBuffer);
280	pReq->pvBounceBuffer = NULL;
281	pReq->cbBounceBuffer = 0;
282	}
283	pReq->fFlags = 0;
284	RTAioMgrFileRelease(pReq->pFile);
285	RTMemCacheFree(pThis->hMemCacheReqs, pReq);
286	}
287
288	static void rtAioMgrReqCompleteRc(PRTAIOMGRINT pThis, PRTAIOMGRREQ pReq,
289	int rcReq, size_t cbTransfered)
290	{
291	int rc = VINF_SUCCESS;
292	PRTAIOMGRFILEINT pFile;
293
294	pFile = pReq->pFile;
295	pThis->cReqsActive--;
296	pFile->AioMgr.cReqsActive--;
297
298	/*
299	* It is possible that the request failed on Linux with kernels < 2.6.23
300	* if the passed buffer was allocated with remap_pfn_range or if the file
301	* is on an NFS endpoint which does not support async and direct I/O at the same time.
302	* The endpoint will be migrated to a failsafe manager in case a request fails.
303	*/
304	if (RT_FAILURE(rcReq))
305	{
306	pFile->pfnReqCompleted(pFile, rcReq, pReq->pvUser);
307	rtAioMgrReqFree(pThis, pReq);
308	}
309	else
310	{
311	/*
312	* Restart an incomplete transfer.
313	* This usually means that the request will return an error now
314	* but to get the cause of the error (disk full, file too big, I/O error, ...)
315	* the transfer needs to be continued.
316	*/
317	if (RT_UNLIKELY( cbTransfered < pReq->DataSeg.cbSeg
318	\|\| ( pReq->cbBounceBuffer
319	&& cbTransfered < pReq->cbBounceBuffer)))
320	{
321	RTFOFF offStart;
322	size_t cbToTransfer;
323	uint8_t *pbBuf = NULL;
324
325	Assert(cbTransfered % 512 == 0);
326
327	if (pReq->cbBounceBuffer)
328	{
329	AssertPtr(pReq->pvBounceBuffer);
330	offStart = (pReq->off & ~((RTFOFF)512-1)) + cbTransfered;
331	cbToTransfer = pReq->cbBounceBuffer - cbTransfered;
332	pbBuf = (uint8_t *)pReq->pvBounceBuffer + cbTransfered;
333	}
334	else
335	{
336	Assert(!pReq->pvBounceBuffer);
337	offStart = pReq->off + cbTransfered;
338	cbToTransfer = pReq->DataSeg.cbSeg - cbTransfered;
339	pbBuf = (uint8_t *)pReq->DataSeg.pvSeg + cbTransfered;
340	}
341
342	if ( pReq->enmType == RTAIOMGRREQTYPE_PREFETCH
343	\|\| pReq->enmType == RTAIOMGRREQTYPE_READ)
344	{
345	rc = RTFileAioReqPrepareRead(pReq->hReqIo, pFile->hFile, offStart,
346	pbBuf, cbToTransfer, pReq);
347	}
348	else
349	{
350	AssertMsg(pReq->enmType == RTAIOMGRREQTYPE_WRITE,
351	("Invalid transfer type\n"));
352	rc = RTFileAioReqPrepareWrite(pReq->hReqIo, pFile->hFile, offStart,
353	pbBuf, cbToTransfer, pReq);
354	}
355	AssertRC(rc);
356
357	rc = rtAioMgrReqsEnqueue(pThis, pFile, &pReq->hReqIo, 1);
358	AssertMsg(RT_SUCCESS(rc) \|\| (rc == VERR_FILE_AIO_INSUFFICIENT_RESSOURCES),
359	("Unexpected return code rc=%Rrc\n", rc));
360	}
361	else if (pReq->enmType == RTAIOMGRREQTYPE_PREFETCH)
362	{
363	Assert(pReq->cbBounceBuffer);
364	pReq->enmType = RTAIOMGRREQTYPE_WRITE;
365
366	memcpy(((uint8_t *)pReq->pvBounceBuffer) + pReq->offBounceBuffer,
367	pReq->DataSeg.pvSeg,
368	pReq->DataSeg.cbSeg);
369
370	/* Write it now. */
371	RTFOFF offStart = pReq->off & ~(RTFOFF)(512-1);
372	size_t cbToTransfer = RT_ALIGN_Z(pReq->DataSeg.cbSeg + (pReq->off - offStart), 512);
373
374	rc = RTFileAioReqPrepareWrite(pReq->hReqIo, pFile->hFile,
375	offStart, pReq->pvBounceBuffer, cbToTransfer, pReq);
376	AssertRC(rc);
377	rc = rtAioMgrReqsEnqueue(pThis, pFile, &pReq->hReqIo, 1);
378	AssertMsg(RT_SUCCESS(rc) \|\| (rc == VERR_FILE_AIO_INSUFFICIENT_RESSOURCES),
379	("Unexpected return code rc=%Rrc\n", rc));
380	}
381	else
382	{
383	if (RT_SUCCESS(rc) && pReq->cbBounceBuffer)
384	{
385	if (pReq->enmType == RTAIOMGRREQTYPE_READ)
386	memcpy(pReq->DataSeg.pvSeg,
387	((uint8_t *)pReq->pvBounceBuffer) + pReq->offBounceBuffer,
388	pReq->DataSeg.cbSeg);
389	}
390
391	/* Call completion callback */
392	pFile->pfnReqCompleted(pFile, rcReq, pReq->pvUser);
393	rtAioMgrReqFree(pThis, pReq);
394	}
395	} /* request completed successfully */
396	}
397
398	/**
399	* Wrapper around rtAioMgrReqCompleteRc().
400	*/
401	static void rtAioMgrReqComplete(PRTAIOMGRINT pThis, RTFILEAIOREQ hReq)
402	{
403	size_t cbTransfered = 0;
404	int rcReq = RTFileAioReqGetRC(hReq, &cbTransfered);
405	PRTAIOMGRREQ pReq = (PRTAIOMGRREQ)RTFileAioReqGetUser(hReq);
406
407	rtAioMgrReqCompleteRc(pThis, pReq, rcReq, cbTransfered);
408	}
409
410	/**
411	* Wrapper around RTFIleAioCtxSubmit() which is also doing error handling.
412	*/
413	static int rtAioMgrReqsEnqueue(PRTAIOMGRINT pThis, PRTAIOMGRFILEINT pFile,
414	PRTFILEAIOREQ pahReqs, unsigned cReqs)
415	{
416	pThis->cReqsActive += cReqs;
417	pFile->AioMgr.cReqsActive += cReqs;
418
419	int rc = RTFileAioCtxSubmit(pThis->hAioCtx, pahReqs, cReqs);
420	if (RT_FAILURE(rc))
421	{
422	if (rc == VERR_FILE_AIO_INSUFFICIENT_RESSOURCES)
423	{
424	/* Append any not submitted task to the waiting list. */
425	for (size_t i = 0; i < cReqs; i++)
426	{
427	int rcReq = RTFileAioReqGetRC(pahReqs[i], NULL);
428
429	if (rcReq != VERR_FILE_AIO_IN_PROGRESS)
430	{
431	PRTAIOMGRREQ pReq = (PRTAIOMGRREQ)RTFileAioReqGetUser(pahReqs[i]);
432
433	Assert(pReq->hReqIo == pahReqs[i]);
434	RTListAppend(&pFile->AioMgr.ListWaitingReqs, &pReq->NodeWaitingList);
435	pThis->cReqsActive--;
436	pFile->AioMgr.cReqsActive--;
437	}
438	}
439
440	pThis->cReqsActiveMax = pThis->cReqsActive;
441	rc = VINF_SUCCESS;
442	}
443	else /* Another kind of error happened (full disk, ...) */
444	{
445	/* An error happened. Find out which one caused the error and resubmit all other tasks. */
446	for (size_t i = 0; i < cReqs; i++)
447	{
448	PRTAIOMGRREQ pReq = (PRTAIOMGRREQ)RTFileAioReqGetUser(pahReqs[i]);
449	int rcReq = RTFileAioReqGetRC(pahReqs[i], NULL);
450
451	if (rcReq == VERR_FILE_AIO_NOT_SUBMITTED)
452	{
453	/* We call ourself again to do any error handling which might come up now. */
454	rc = rtAioMgrReqsEnqueue(pThis, pFile, &pahReqs[i], 1);
455	AssertRC(rc);
456	}
457	else if (rcReq != VERR_FILE_AIO_IN_PROGRESS)
458	rtAioMgrReqCompleteRc(pThis, pReq, rcReq, 0);
459	}
460	}
461	}
462
463	return VINF_SUCCESS;
464	}
465
466	/**
467	* Adds a list of requests to the waiting list.
468	*
469	* @returns nothing.
470	* @param pFile The file instance to add the requests to.
471	* @param pReqsHead The head of the request list to add.
472	*/
473	static void rtAioMgrFileAddReqsToWaitingList(PRTAIOMGRFILEINT pFile, PRTAIOMGRREQ pReqsHead)
474	{
475	while (pReqsHead)
476	{
477	PRTAIOMGRREQ pReqCur = pReqsHead;
478
479	pReqsHead = (PRTAIOMGRREQ)pReqsHead->WorkItem.pNext;
480	pReqCur->WorkItem.pNext = NULL;
481	RTListAppend(&pFile->AioMgr.ListWaitingReqs, &pReqCur->NodeWaitingList);
482	}
483	}
484
485	/**
486	* Prepare the native I/o request ensuring that all alignment prerequisites of
487	* the host are met.
488	*
489	* @returns IPRT statuse code.
490	* @param pFile The file instance data.
491	* @param pReq The request to prepare.
492	*/
493	static int rtAioMgrReqPrepareNonBuffered(PRTAIOMGRFILEINT pFile, PRTAIOMGRREQ pReq)
494	{
495	int rc = VINF_SUCCESS;
496	RTFOFF offStart = pReq->off & ~(RTFOFF)(512-1);
497	size_t cbToTransfer = RT_ALIGN_Z(pReq->DataSeg.cbSeg + (pReq->off - offStart), 512);
498	void *pvBuf = pReq->DataSeg.pvSeg;
499	bool fAlignedReq = cbToTransfer == pReq->DataSeg.cbSeg
500	&& offStart == pReq->off;
501
502	/*
503	* Check if the alignment requirements are met.
504	* Offset, transfer size and buffer address
505	* need to be on a 512 boundary.
506	*/
507	if ( !fAlignedReq
508	/** @todo: \|\| ((pEpClassFile->uBitmaskAlignment & (RTR3UINTPTR)pvBuf) != (RTR3UINTPTR)pvBuf) */)
509	{
510	/* Create bounce buffer. */
511	pReq->cbBounceBuffer = cbToTransfer;
512
513	AssertMsg(pReq->off >= offStart, ("Overflow in calculation off=%llu offStart=%llu\n",
514	pReq->off, offStart));
515	pReq->offBounceBuffer = pReq->off - offStart;
516
517	/** @todo: I think we need something like a RTMemAllocAligned method here.
518	* Current assumption is that the maximum alignment is 4096byte
519	* (GPT disk on Windows)
520	* so we can use RTMemPageAlloc here.
521	*/
522	pReq->pvBounceBuffer = RTMemPageAlloc(cbToTransfer);
523	if (RT_LIKELY(pReq->pvBounceBuffer))
524	{
525	pvBuf = pReq->pvBounceBuffer;
526
527	if (pReq->enmType == RTAIOMGRREQTYPE_WRITE)
528	{
529	if ( RT_UNLIKELY(cbToTransfer != pReq->DataSeg.cbSeg)
530	\|\| RT_UNLIKELY(offStart != pReq->off))
531	{
532	/* We have to fill the buffer first before we can update the data. */
533	pReq->enmType = RTAIOMGRREQTYPE_WRITE;
534	}
535	else
536	memcpy(pvBuf, pReq->DataSeg.pvSeg, pReq->DataSeg.cbSeg);
537	}
538	}
539	else
540	rc = VERR_NO_MEMORY;
541	}
542	else
543	pReq->cbBounceBuffer = 0;
544
545	if (RT_SUCCESS(rc))
546	{
547	if (pReq->enmType == RTAIOMGRREQTYPE_WRITE)
548	{
549	rc = RTFileAioReqPrepareWrite(pReq->hReqIo, pFile->hFile,
550	offStart, pvBuf, cbToTransfer, pReq);
551	}
552	else /* Read or prefetch request. */
553	rc = RTFileAioReqPrepareRead(pReq->hReqIo, pFile->hFile,
554	offStart, pvBuf, cbToTransfer, pReq);
555	AssertRC(rc);
556	pReq->fFlags \|= RTAIOMGRREQ_FLAGS_PREPARED;
557	}
558
559	return rc;
560	}
561
562	/**
563	* Prepare a new request for enqueuing.
564	*
565	* @returns IPRT status code.
566	* @param pReq The request to prepare.
567	* @param phReqIo Where to store the handle to the native I/O request on success.
568	*/
569	static int rtAioMgrPrepareReq(PRTAIOMGRREQ pReq, PRTFILEAIOREQ phReqIo)
570	{
571	int rc = VINF_SUCCESS;
572	PRTAIOMGRFILEINT pFile = pReq->pFile;
573
574	switch (pReq->enmType)
575	{
576	case RTAIOMGRREQTYPE_FLUSH:
577	{
578	rc = RTFileAioReqPrepareFlush(pReq->hReqIo, pFile->hFile, pReq);
579	break;
580	}
581	case RTAIOMGRREQTYPE_READ:
582	case RTAIOMGRREQTYPE_WRITE:
583	{
584	rc = rtAioMgrReqPrepareNonBuffered(pFile, pReq);
585	break;
586	}
587	default:
588	AssertMsgFailed(("Invalid transfer type %d\n", pReq->enmType));
589	} /* switch transfer type */
590
591	if (RT_SUCCESS(rc))
592	*phReqIo = pReq->hReqIo;
593
594	return rc;
595	}
596
597	/**
598	* Prepare newly submitted requests for processing.
599	*
600	* @returns IPRT status code
601	* @param pThis The async I/O manager instance data.
602	* @param pFile The file instance.
603	* @param pReqsNew The list of new requests to prepare.
604	*/
605	static int rtAioMgrPrepareNewReqs(PRTAIOMGRINT pThis,
606	PRTAIOMGRFILEINT pFile,
607	PRTAIOMGRREQ pReqsNew)
608	{
609	RTFILEAIOREQ apReqs[20];
610	unsigned cRequests = 0;
611	int rc = VINF_SUCCESS;
612
613	/* Go through the list and queue the requests. */
614	while ( pReqsNew
615	&& (pThis->cReqsActive + cRequests < pThis->cReqsActiveMax)
616	&& RT_SUCCESS(rc))
617	{
618	PRTAIOMGRREQ pCurr = pReqsNew;
619	pReqsNew = (PRTAIOMGRREQ)pReqsNew->WorkItem.pNext;
620
621	pCurr->WorkItem.pNext = NULL;
622	AssertMsg(VALID_PTR(pCurr->pFile) && (pCurr->pFile == pFile),
623	("Files do not match\n"));
624	AssertMsg(!(pCurr->fFlags & RTAIOMGRREQ_FLAGS_PREPARED),
625	("Request on the new list is already prepared\n"));
626
627	rc = rtAioMgrPrepareReq(pCurr, &apReqs[cRequests]);
628	if (RT_FAILURE(rc))
629	rtAioMgrReqCompleteRc(pThis, pCurr, rc, 0);
630	else
631	cRequests++;
632
633	/* Queue the requests if the array is full. */
634	if (cRequests == RT_ELEMENTS(apReqs))
635	{
636	rc = rtAioMgrReqsEnqueue(pThis, pFile, apReqs, cRequests);
637	cRequests = 0;
638	AssertMsg(RT_SUCCESS(rc) \|\| (rc == VERR_FILE_AIO_INSUFFICIENT_RESSOURCES),
639	("Unexpected return code\n"));
640	}
641	}
642
643	if (cRequests)
644	{
645	rc = rtAioMgrReqsEnqueue(pThis, pFile, apReqs, cRequests);
646	AssertMsg(RT_SUCCESS(rc) \|\| (rc == VERR_FILE_AIO_INSUFFICIENT_RESSOURCES),
647	("Unexpected return code rc=%Rrc\n", rc));
648	}
649
650	if (pReqsNew)
651	{
652	/* Add the rest of the tasks to the pending list */
653	rtAioMgrFileAddReqsToWaitingList(pFile, pReqsNew);
654	}
655
656	/* Insufficient resources are not fatal. */
657	if (rc == VERR_FILE_AIO_INSUFFICIENT_RESSOURCES)
658	rc = VINF_SUCCESS;
659
660	return rc;
661	}
662
663	/**
664	* Queues waiting requests.
665	*
666	* @returns IPRT status code.
667	* @param pThis The async I/O manager instance data.
668	* @param pFile The file to get the requests from.
669	*/
670	static int rtAioMgrQueueWaitingReqs(PRTAIOMGRINT pThis, PRTAIOMGRFILEINT pFile)
671	{
672	RTFILEAIOREQ apReqs[20];
673	unsigned cRequests = 0;
674	int rc = VINF_SUCCESS;
675	PRTAIOMGRREQ pReqIt;
676	PRTAIOMGRREQ pReqItNext;
677
678	/* Go through the list and queue the requests. */
679	RTListForEachSafe(&pFile->AioMgr.ListWaitingReqs, pReqIt, pReqItNext, RTAIOMGRREQ, NodeWaitingList)
680	{
681	RTListNodeRemove(&pReqIt->NodeWaitingList);
682	AssertMsg(VALID_PTR(pReqIt->pFile) && (pReqIt->pFile == pFile),
683	("Files do not match\n"));
684
685	if (!(pReqIt->fFlags & RTAIOMGRREQ_FLAGS_PREPARED))
686	{
687	rc = rtAioMgrPrepareReq(pReqIt, &apReqs[cRequests]);
688	if (RT_FAILURE(rc))
689	rtAioMgrReqCompleteRc(pThis, pReqIt, rc, 0);
690	else
691	cRequests++;
692	}
693	else
694	{
695	apReqs[cRequests] = pReqIt->hReqIo;
696	cRequests++;
697	}
698
699	/* Queue the requests if the array is full. */
700	if (cRequests == RT_ELEMENTS(apReqs))
701	{
702	rc = rtAioMgrReqsEnqueue(pThis, pFile, apReqs, cRequests);
703	cRequests = 0;
704	AssertMsg(RT_SUCCESS(rc) \|\| (rc == VERR_FILE_AIO_INSUFFICIENT_RESSOURCES),
705	("Unexpected return code\n"));
706	}
707	}
708
709	if (cRequests)
710	{
711	rc = rtAioMgrReqsEnqueue(pThis, pFile, apReqs, cRequests);
712	AssertMsg(RT_SUCCESS(rc) \|\| (rc == VERR_FILE_AIO_INSUFFICIENT_RESSOURCES),
713	("Unexpected return code rc=%Rrc\n", rc));
714	}
715
716	/* Insufficient resources are not fatal. */
717	if (rc == VERR_FILE_AIO_INSUFFICIENT_RESSOURCES)
718	rc = VINF_SUCCESS;
719
720	return rc;
721	}
722
723	/**
724	* Adds all pending requests for the given file.
725	*
726	* @returns IPRT status code.
727	* @param pThis The async I/O manager instance data.
728	* @param pFile The file to get the requests from.
729	*/
730	static int rtAioMgrQueueReqs(PRTAIOMGRINT pThis, PRTAIOMGRFILEINT pFile)
731	{
732	int rc = VINF_SUCCESS;
733	PRTAIOMGRFILEINT pReqsHead = NULL;
734
735	/* Check the pending list first */
736	if (!RTListIsEmpty(&pFile->AioMgr.ListWaitingReqs))
737	rc = rtAioMgrQueueWaitingReqs(pThis, pFile);
738
739	if ( RT_SUCCESS(rc)
740	&& RTListIsEmpty(&pFile->AioMgr.ListWaitingReqs))
741	{
742	PRTAIOMGRREQ pReqsNew = (PRTAIOMGRREQ)RTQueueAtomicRemoveAll(&pFile->QueueReqs);
743
744	if (pReqsNew)
745	{
746	rc = rtAioMgrPrepareNewReqs(pThis, pFile, pReqsNew);
747	AssertRC(rc);
748	}
749	}
750
751	return rc;
752	}
753
754	/**
755	* Checks all files for new requests.
756	*
757	* @returns IPRT status code.
758	* @param pThis The I/O manager instance data.
759	*/
760	static int rtAioMgrCheckFiles(PRTAIOMGRINT pThis)
761	{
762	int rc = VINF_SUCCESS;
763	PRTAIOMGRFILEINT pIt;
764
765	RTListForEach(&pThis->ListFiles, pIt, RTAIOMGRFILEINT, AioMgr.NodeAioMgrFiles)
766	{
767	rc = rtAioMgrQueueReqs(pThis, pIt);
768	if (RT_FAILURE(rc))
769	return rc;
770	}
771
772	return rc;
773	}
774
775	/**
776	* Process a blocking event from the outside.
777	*
778	* @returns IPRT status code.
779	* @param pThis The async I/O manager instance data.
780	*/
781	static int rtAioMgrProcessBlockingEvent(PRTAIOMGRINT pThis)
782	{
783	int rc = VINF_SUCCESS;
784	bool fNotifyWaiter = false;
785
786	switch (pThis->enmBlockingEvent)
787	{
788	case RTAIOMGREVENT_NO_EVENT:
789	/* Nothing to do. */
790	break;
791	case RTAIOMGREVENT_FILE_ADD:
792	{
793	PRTAIOMGRFILEINT pFile = ASMAtomicReadPtrT(&pThis->BlockingEventData.pFileAdd, PRTAIOMGRFILEINT);
794	AssertMsg(VALID_PTR(pFile), ("Adding file event without a file to add\n"));
795
796	RTListAppend(&pThis->ListFiles, &pFile->AioMgr.NodeAioMgrFiles);
797	fNotifyWaiter = true;
798	break;
799	}
800	case RTAIOMGREVENT_FILE_CLOSE:
801	{
802	PRTAIOMGRFILEINT pFile = ASMAtomicReadPtrT(&pThis->BlockingEventData.pFileClose, PRTAIOMGRFILEINT);
803	AssertMsg(VALID_PTR(pFile), ("Close file event without a file to close\n"));
804
805	if (!(pFile->fFlags & RTAIOMGRFILE_FLAGS_CLOSING))
806	{
807	/* Make sure all requests finished. Process the queues a last time first. */
808	rc = rtAioMgrQueueReqs(pThis, pFile);
809	AssertRC(rc);
810
811	pFile->fFlags \|= RTAIOMGRFILE_FLAGS_CLOSING;
812	fNotifyWaiter = !rtAioMgrFileRemove(pFile);
813	}
814	else if (!pFile->AioMgr.cReqsActive)
815	fNotifyWaiter = true;
816	break;
817	}
818	case RTAIOMGREVENT_SHUTDOWN:
819	{
820	if (!pThis->cReqsActive)
821	fNotifyWaiter = true;
822	break;
823	}
824	default:
825	AssertReleaseMsgFailed(("Invalid event type %d\n", pThis->enmBlockingEvent));
826	}
827
828	if (fNotifyWaiter)
829	{
830	/* Release the waiting thread. */
831	rc = RTSemEventSignal(pThis->hEventSemBlock);
832	AssertRC(rc);
833	}
834
835	return rc;
836	}
837
838	/**
839	* async I/O manager worker loop.
840	*
841	* @returns IPRT status code.
842	* @param hThreadSelf The thread handle this worker belongs to.
843	* @param pvUser Opaque user data (Pointer to async I/O manager instance).
844	*/
845	static DECLCALLBACK(int) rtAioMgrWorker(RTTHREAD hThreadSelf, void *pvUser)
846	{
847	PRTAIOMGRINT pThis = (PRTAIOMGRINT)pvUser;
848	bool fRunning = true;
849	int rc = VINF_SUCCESS;
850
851	do
852	{
853	uint32_t cReqsCompleted = 0;
854	RTFILEAIOREQ ahReqsCompleted[32];
855	rc = RTFileAioCtxWait(pThis->hAioCtx, 1, RT_INDEFINITE_WAIT, &ahReqsCompleted[0],
856	RT_ELEMENTS(ahReqsCompleted), &cReqsCompleted);
857	if (rc == VERR_INTERRUPTED)
858	{
859	/* Process external event. */
860	rtAioMgrProcessBlockingEvent(pThis);
861	rc = rtAioMgrCheckFiles(pThis);
862	}
863	else if (RT_FAILURE(rc))
864	{
865	/* Something bad happened. */
866	/** @todo: */
867	}
868	else
869	{
870	/* Requests completed. */
871	for (uint32_t i = 0; i < cReqsCompleted; i++)
872	rtAioMgrReqComplete(pThis, ahReqsCompleted[i]);
873
874	/* Check files for new requests and queue waiting requests. */
875	rc = rtAioMgrCheckFiles(pThis);
876	}
877	} while ( fRunning
878	&& RT_SUCCESS(rc));
879
880	return rc;
881	}
882
883	/**
884	* Wakes up the async I/O manager.
885	*
886	* @returns IPRT status code.
887	* @param pThis The async I/O manager.
888	*/
889	static int rtAioMgrWakeup(PRTAIOMGRINT pThis)
890	{
891	return RTFileAioCtxWakeup(pThis->hAioCtx);
892	}
893
894	/**
895	* Waits until the async I/O manager handled the given event.
896	*
897	* @returns IPRT status code.
898	* @param pThis The async I/O manager.
899	* @param enmEvent The event to pass to the manager.
900	*/
901	static int rtAioMgrWaitForBlockingEvent(PRTAIOMGRINT pThis, RTAIOMGREVENT enmEvent)
902	{
903	Assert(pThis->enmBlockingEvent == RTAIOMGREVENT_NO_EVENT);
904	ASMAtomicWriteU32((volatile uint32_t *)&pThis->enmBlockingEvent, enmEvent);
905
906	/* Wakeup the async I/O manager */
907	int rc = rtAioMgrWakeup(pThis);
908	if (RT_FAILURE(rc))
909	return rc;
910
911	/* Wait for completion. */
912	rc = RTSemEventWait(pThis->hEventSemBlock, RT_INDEFINITE_WAIT);
913	AssertRC(rc);
914
915	ASMAtomicWriteU32((volatile uint32_t *)&pThis->enmBlockingEvent, RTAIOMGREVENT_NO_EVENT);
916
917	return rc;
918	}
919
920	/**
921	* Add a given file to the given I/O manager.
922	*
923	* @returns IPRT status code.
924	* @param pThis The async I/O manager.
925	* @param pFile The file to add.
926	*/
927	static int rtAioMgrAddFile(PRTAIOMGRINT pThis, PRTAIOMGRFILEINT pFile)
928	{
929	/* Update the assigned I/O manager. */
930	ASMAtomicWritePtr(&pFile->pAioMgr, pThis);
931
932	int rc = RTCritSectEnter(&pThis->CritSectBlockingEvent);
933	AssertRCReturn(rc, rc);
934
935	ASMAtomicWritePtr(&pThis->BlockingEventData.pFileAdd, pFile);
936	rc = rtAioMgrWaitForBlockingEvent(pThis, RTAIOMGREVENT_FILE_ADD);
937	ASMAtomicWriteNullPtr(&pThis->BlockingEventData.pFileAdd);
938
939	RTCritSectLeave(&pThis->CritSectBlockingEvent);
940	return rc;
941	}
942
943	/**
944	* Removes a given file from the given I/O manager.
945	*
946	* @returns IPRT status code.
947	* @param pThis The async I/O manager.
948	* @param pFile The file to remove.
949	*/
950	static int rtAioMgrCloseFile(PRTAIOMGRINT pThis, PRTAIOMGRFILEINT pFile)
951	{
952	int rc = RTCritSectEnter(&pThis->CritSectBlockingEvent);
953	AssertRCReturn(rc, rc);
954
955	ASMAtomicWritePtr(&pThis->BlockingEventData.pFileClose, pFile);
956	rc = rtAioMgrWaitForBlockingEvent(pThis, RTAIOMGREVENT_FILE_CLOSE);
957	ASMAtomicWriteNullPtr(&pThis->BlockingEventData.pFileClose);
958
959	RTCritSectLeave(&pThis->CritSectBlockingEvent);
960
961	return rc;
962	}
963
964	/**
965	* Process a shutdown event.
966	*
967	* @returns IPRT status code.
968	* @param pThis The async I/O manager to shut down.
969	*/
970	static int rtAioMgrShutdown(PRTAIOMGRINT pThis)
971	{
972	int rc = RTCritSectEnter(&pThis->CritSectBlockingEvent);
973	AssertRCReturn(rc, rc);
974
975	rc = rtAioMgrWaitForBlockingEvent(pThis, RTAIOMGREVENT_SHUTDOWN);
976	RTCritSectLeave(&pThis->CritSectBlockingEvent);
977
978	return rc;
979	}
980
981	/**
982	* Destroys an async I/O manager.
983	*
984	* @returns nothing.
985	* @param pThis The async I/O manager instance to destroy.
986	*/
987	static void rtAioMgrDestroy(PRTAIOMGRINT pThis)
988	{
989	int rc;
990
991	rc = rtAioMgrShutdown(pThis);
992	AssertRC(rc);
993
994	rc = RTThreadWait(pThis->hThread, RT_INDEFINITE_WAIT, NULL);
995	AssertRC(rc);
996
997	rc = RTFileAioCtxDestroy(pThis->hAioCtx);
998	AssertRC(rc);
999
1000	rc = RTMemCacheDestroy(pThis->hMemCacheReqs);
1001	AssertRC(rc);
1002
1003	pThis->hThread = NIL_RTTHREAD;
1004	pThis->hAioCtx = NIL_RTFILEAIOCTX;
1005	pThis->hMemCacheReqs = NIL_RTMEMCACHE;
1006	pThis->u32Magic = ~RTAIOMGR_MAGIC;
1007	RTCritSectDelete(&pThis->CritSectBlockingEvent);
1008	RTSemEventDestroy(pThis->hEventSemBlock);
1009	RTMemFree(pThis);
1010	}
1011
1012	/**
1013	* Queues a new request for processing.
1014	*/
1015	static void rtAioMgrFileQueueReq(PRTAIOMGRFILEINT pThis, PRTAIOMGRREQ pReq)
1016	{
1017	RTAioMgrFileRetain(pThis);
1018	RTQueueAtomicInsert(&pThis->QueueReqs, &pReq->WorkItem);
1019	rtAioMgrWakeup(pThis->pAioMgr);
1020	}
1021
1022	/**
1023	* Destroys an async I/O manager file.
1024	*
1025	* @returns nothing.
1026	* @param pThis The async I/O manager file.
1027	*/
1028	static void rtAioMgrFileDestroy(PRTAIOMGRFILEINT pThis)
1029	{
1030	pThis->u32Magic = ~RTAIOMGRFILE_MAGIC;
1031	rtAioMgrCloseFile(pThis->pAioMgr, pThis);
1032	RTAioMgrRelease(pThis->pAioMgr);
1033	RTMemFree(pThis);
1034	}
1035
1036	/**
1037	* Queues a new I/O request.
1038	*
1039	* @returns IPRT status code.
1040	* @param hAioMgrFile The I/O manager file handle.
1041	* @param off Start offset of the I/o request.
1042	* @param pSgBuf Data S/G buffer.
1043	* @param cbIo How much to transfer.
1044	* @param pvUser Opaque user data.
1045	* @param enmType I/O direction type (read/write).
1046	*/
1047	static int rtAioMgrFileIoReqCreate(RTAIOMGRFILE hAioMgrFile, RTFOFF off, PRTSGBUF pSgBuf,
1048	size_t cbIo, void *pvUser, RTAIOMGRREQTYPE enmType)
1049	{
1050	int rc;
1051	PRTAIOMGRFILEINT pFile = hAioMgrFile;
1052	PRTAIOMGRINT pAioMgr;
1053
1054	AssertPtrReturn(pFile, VERR_INVALID_HANDLE);
1055	pAioMgr = pFile->pAioMgr;
1056
1057	PRTAIOMGRREQ pReq = rtAioMgrReqAlloc(pAioMgr);
1058	if (RT_LIKELY(pReq))
1059	{
1060	unsigned cSeg = 1;
1061	size_t cbSeg = RTSgBufSegArrayCreate(pSgBuf, &pReq->DataSeg, &cSeg, cbIo);
1062
1063	if (cbSeg == cbIo)
1064	{
1065	pReq->enmType = enmType;
1066	pReq->pFile = pFile;
1067	pReq->pvUser = pvUser;
1068	pReq->off = off;
1069	rtAioMgrFileQueueReq(pFile, pReq);
1070	rc = VERR_FILE_AIO_IN_PROGRESS;
1071	}
1072	else
1073	{
1074	/** @todo: Real S/G buffer support. */
1075	rtAioMgrReqFree(pAioMgr, pReq);
1076	rc = VERR_NOT_SUPPORTED;
1077	}
1078	}
1079	else
1080	rc = VERR_NO_MEMORY;
1081
1082	return rc;
1083	}
1084
1085	/**
1086	* Request constructor for the memory cache.
1087	*
1088	* @returns IPRT status code.
1089	* @param hMemCache The cache handle.
1090	* @param pvObj The memory object that should be initialized.
1091	* @param pvUser The user argument.
1092	*/
1093	static DECLCALLBACK(int) rtAioMgrReqCtor(RTMEMCACHE hMemCache, void pvObj, void pvUser)
1094	{
1095	PRTAIOMGRREQ pReq = (PRTAIOMGRREQ)pvObj;
1096
1097	memset(pReq, 0, sizeof(RTAIOMGRREQ));
1098	return RTFileAioReqCreate(&pReq->hReqIo);
1099	}
1100
1101	/**
1102	* Request destructor for the memory cache.
1103	*
1104	* @param hMemCache The cache handle.
1105	* @param pvObj The memory object that should be destroyed.
1106	* @param pvUser The user argument.
1107	*/
1108	static DECLCALLBACK(void) rtAioMgrReqDtor(RTMEMCACHE hMemCache, void pvObj, void pvUser)
1109	{
1110	PRTAIOMGRREQ pReq = (PRTAIOMGRREQ)pvObj;
1111	int rc = RTFileAioReqDestroy(pReq->hReqIo);
1112
1113	AssertRC(rc);
1114	}
1115
1116	RTDECL(int) RTAioMgrCreate(PRTAIOMGR phAioMgr, uint32_t cReqsMax)
1117	{
1118	int rc = VINF_SUCCESS;
1119	PRTAIOMGRINT pThis;
1120
1121	AssertPtrReturn(phAioMgr, VERR_INVALID_POINTER);
1122	AssertReturn(cReqsMax > 0, VERR_INVALID_PARAMETER);
1123
1124	pThis = (PRTAIOMGRINT)RTMemAllocZ(sizeof(RTAIOMGRINT));
1125	if (pThis)
1126	{
1127	pThis->u32Magic = RTAIOMGR_MAGIC;
1128	pThis->cRefs = 1;
1129	pThis->enmBlockingEvent = RTAIOMGREVENT_NO_EVENT;
1130	RTListInit(&pThis->ListFiles);
1131	rc = RTCritSectInit(&pThis->CritSectBlockingEvent);
1132	if (RT_SUCCESS(rc))
1133	{
1134	rc = RTSemEventCreate(&pThis->hEventSemBlock);
1135	if (RT_SUCCESS(rc))
1136	{
1137	rc = RTMemCacheCreate(&pThis->hMemCacheReqs, sizeof(RTAIOMGRREQ),
1138	0, UINT32_MAX, rtAioMgrReqCtor, rtAioMgrReqDtor, NULL, 0);
1139	if (RT_SUCCESS(rc))
1140	{
1141	rc = RTFileAioCtxCreate(&pThis->hAioCtx, cReqsMax == UINT32_MAX
1142	? RTFILEAIO_UNLIMITED_REQS
1143	: cReqsMax,
1144	RTFILEAIOCTX_FLAGS_WAIT_WITHOUT_PENDING_REQUESTS);
1145	if (RT_SUCCESS(rc))
1146	{
1147	rc = RTThreadCreateF(&pThis->hThread, rtAioMgrWorker, pThis, 0, RTTHREADTYPE_IO,
1148	RTTHREADFLAGS_WAITABLE, "AioMgr-%u", cReqsMax);
1149	if (RT_FAILURE(rc))
1150	{
1151	rc = RTFileAioCtxDestroy(pThis->hAioCtx);
1152	AssertRC(rc);
1153	}
1154	}
1155
1156	if (RT_FAILURE(rc))
1157	RTMemCacheDestroy(pThis->hMemCacheReqs);
1158	}
1159
1160	if (RT_FAILURE(rc))
1161	RTSemEventDestroy(pThis->hEventSemBlock);
1162	}
1163
1164	if (RT_FAILURE(rc))
1165	RTCritSectDelete(&pThis->CritSectBlockingEvent);
1166	}
1167
1168	if (RT_FAILURE(rc))
1169	RTMemFree(pThis);
1170	}
1171	else
1172	rc = VERR_NO_MEMORY;
1173
1174	if (RT_SUCCESS(rc))
1175	*phAioMgr = pThis;
1176
1177	return rc;
1178	}
1179
1180	RTDECL(uint32_t) RTAioMgrRetain(RTAIOMGR hAioMgr)
1181	{
1182	PRTAIOMGRINT pThis = hAioMgr;
1183	AssertReturn(hAioMgr != NIL_RTAIOMGR, UINT32_MAX);
1184	AssertPtrReturn(pThis, UINT32_MAX);
1185	AssertReturn(pThis->u32Magic == RTAIOMGR_MAGIC, UINT32_MAX);
1186
1187	uint32_t cRefs = ASMAtomicIncU32(&pThis->cRefs);
1188	AssertMsg(cRefs > 1 && cRefs < _1G, ("%#x %p\n", cRefs, pThis));
1189	return cRefs;
1190	}
1191
1192	RTDECL(uint32_t) RTAioMgrRelease(RTAIOMGR hAioMgr)
1193	{
1194	PRTAIOMGRINT pThis = hAioMgr;
1195	if (pThis == NIL_RTAIOMGR)
1196	return 0;
1197	AssertPtrReturn(pThis, UINT32_MAX);
1198	AssertReturn(pThis->u32Magic == RTAIOMGR_MAGIC, UINT32_MAX);
1199
1200	uint32_t cRefs = ASMAtomicDecU32(&pThis->cRefs);
1201	AssertMsg(cRefs < _1G, ("%#x %p\n", cRefs, pThis));
1202	if (cRefs == 0)
1203	rtAioMgrDestroy(pThis);
1204	return cRefs;
1205	}
1206
1207	RTDECL(int) RTAioMgrFileCreate(RTAIOMGR hAioMgr, RTFILE hFile, PFNRTAIOMGRREQCOMPLETE pfnReqComplete,
1208	void *pvUser, PRTAIOMGRFILE phAioMgrFile)
1209	{
1210	int rc = VINF_SUCCESS;
1211	PRTAIOMGRFILEINT pThis;
1212
1213	AssertReturn(hAioMgr != NIL_RTAIOMGR, VERR_INVALID_HANDLE);
1214	AssertReturn(hFile != NIL_RTFILE, VERR_INVALID_HANDLE);
1215	AssertPtrReturn(pfnReqComplete, VERR_INVALID_POINTER);
1216	AssertPtrReturn(phAioMgrFile, VERR_INVALID_POINTER);
1217
1218	pThis = (PRTAIOMGRFILEINT)RTMemAllocZ(sizeof(RTAIOMGRFILEINT));
1219	if (pThis)
1220	{
1221	pThis->u32Magic = RTAIOMGRFILE_MAGIC;
1222	pThis->cRefs = 1;
1223	pThis->hFile = hFile;
1224	pThis->pAioMgr = hAioMgr;
1225	pThis->pvUser = pvUser;
1226	pThis->pfnReqCompleted = pfnReqComplete;
1227	RTQueueAtomicInit(&pThis->QueueReqs);
1228	RTListInit(&pThis->AioMgr.ListWaitingReqs);
1229	RTAioMgrRetain(hAioMgr);
1230	rc = RTFileAioCtxAssociateWithFile(pThis->pAioMgr->hAioCtx, hFile);
1231	if (RT_FAILURE(rc))
1232	rtAioMgrFileDestroy(pThis);
1233	else
1234	rtAioMgrAddFile(pThis->pAioMgr, pThis);
1235	}
1236	else
1237	rc = VERR_NO_MEMORY;
1238
1239	if (RT_SUCCESS(rc))
1240	*phAioMgrFile = pThis;
1241
1242	return rc;
1243	}
1244
1245	RTDECL(uint32_t) RTAioMgrFileRetain(RTAIOMGRFILE hAioMgrFile)
1246	{
1247	PRTAIOMGRFILEINT pThis = hAioMgrFile;
1248	AssertReturn(hAioMgrFile != NIL_RTAIOMGRFILE, UINT32_MAX);
1249	AssertPtrReturn(pThis, UINT32_MAX);
1250	AssertReturn(pThis->u32Magic == RTAIOMGRFILE_MAGIC, UINT32_MAX);
1251
1252	uint32_t cRefs = ASMAtomicIncU32(&pThis->cRefs);
1253	AssertMsg(cRefs > 1 && cRefs < _1G, ("%#x %p\n", cRefs, pThis));
1254	return cRefs;
1255	}
1256
1257	RTDECL(uint32_t) RTAioMgrFileRelease(RTAIOMGRFILE hAioMgrFile)
1258	{
1259	PRTAIOMGRFILEINT pThis = hAioMgrFile;
1260	if (pThis == NIL_RTAIOMGRFILE)
1261	return 0;
1262	AssertPtrReturn(pThis, UINT32_MAX);
1263	AssertReturn(pThis->u32Magic == RTAIOMGRFILE_MAGIC, UINT32_MAX);
1264
1265	uint32_t cRefs = ASMAtomicDecU32(&pThis->cRefs);
1266	AssertMsg(cRefs < _1G, ("%#x %p\n", cRefs, pThis));
1267	if (cRefs == 0)
1268	rtAioMgrFileDestroy(pThis);
1269	return cRefs;
1270	}
1271
1272	RTDECL(void *) RTAioMgrFileGetUser(RTAIOMGRFILE hAioMgrFile)
1273	{
1274	PRTAIOMGRFILEINT pThis = hAioMgrFile;
1275
1276	AssertPtrReturn(pThis, NULL);
1277	return pThis->pvUser;
1278	}
1279
1280	RTDECL(int) RTAioMgrFileRead(RTAIOMGRFILE hAioMgrFile, RTFOFF off,
1281	PRTSGBUF pSgBuf, size_t cbRead, void *pvUser)
1282	{
1283	return rtAioMgrFileIoReqCreate(hAioMgrFile, off, pSgBuf, cbRead, pvUser,
1284	RTAIOMGRREQTYPE_READ);
1285	}
1286
1287	RTDECL(int) RTAioMgrFileWrite(RTAIOMGRFILE hAioMgrFile, RTFOFF off,
1288	PRTSGBUF pSgBuf, size_t cbWrite, void *pvUser)
1289	{
1290	return rtAioMgrFileIoReqCreate(hAioMgrFile, off, pSgBuf, cbWrite, pvUser,
1291	RTAIOMGRREQTYPE_WRITE);
1292	}
1293
1294	RTDECL(int) RTAioMgrFileFlush(RTAIOMGRFILE hAioMgrFile, void *pvUser)
1295	{
1296	PRTAIOMGRFILEINT pFile = hAioMgrFile;
1297	PRTAIOMGRINT pAioMgr;
1298
1299	AssertPtrReturn(pFile, VERR_INVALID_HANDLE);
1300
1301	pAioMgr = pFile->pAioMgr;
1302
1303	PRTAIOMGRREQ pReq = rtAioMgrReqAlloc(pAioMgr);
1304	if (RT_UNLIKELY(!pReq))
1305	return VERR_NO_MEMORY;
1306
1307	pReq->pFile = pFile;
1308	pReq->enmType = RTAIOMGRREQTYPE_FLUSH;
1309	pReq->pvUser = pvUser;
1310	rtAioMgrFileQueueReq(pFile, pReq);
1311
1312	return VERR_FILE_AIO_IN_PROGRESS;
1313	}
1314

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

source: vbox/trunk/src/VBox/Runtime/common/misc/aiomgr.cpp@ 49329

Download in other formats: