VirtualBox

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

Last change on this file since 56290 was 56290, checked in by vboxsync, 9 years ago

IPRT: Updated (C) year.

  • 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 56290 2015-06-09 14:01:31Z vboxsync $ */
2/** @file
3 * IPRT - Async I/O manager.
4 */
5
6/*
7 * Copyright (C) 2013-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* 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. */
52typedef struct RTAIOMGRFILEINT *PRTAIOMGRFILEINT;
53
54/**
55 * Blocking event types.
56 */
57typedef 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 */
76typedef 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. */
112typedef RTAIOMGRINT *PRTAIOMGRINT;
113
114/**
115 * Async I/O manager file instance data.
116 */
117typedef 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 */
154typedef 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. */
170typedef RTAIOMGRREQTYPE *PRTAIOMGRREQTYPE;
171
172/**
173 * Async I/O manager request.
174 */
175typedef 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. */
204typedef 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
234static 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 */
244static 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 */
262static 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 */
274static 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
288static 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 */
401static 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 */
413static 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 */
473static 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 */
493static 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 */
569static 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 */
605static 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 */
670static 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 */
730static 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 */
760static 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 */
781static 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 */
845static 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 */
889static 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 */
901static 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 */
927static 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 */
950static 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 */
970static 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 */
987static 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 */
1015static 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 */
1028static 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 */
1047static 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 */
1093static 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 */
1108static 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
1116RTDECL(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
1180RTDECL(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
1192RTDECL(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
1207RTDECL(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
1245RTDECL(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
1257RTDECL(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
1272RTDECL(void *) RTAioMgrFileGetUser(RTAIOMGRFILE hAioMgrFile)
1273{
1274 PRTAIOMGRFILEINT pThis = hAioMgrFile;
1275
1276 AssertPtrReturn(pThis, NULL);
1277 return pThis->pvUser;
1278}
1279
1280RTDECL(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
1287RTDECL(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
1294RTDECL(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.

© 2024 Oracle Support Privacy / Do Not Sell My Info Terms of Use Trademark Policy Automated Access Etiquette