VirtualBox

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

Last change on this file since 31453 was 30112, checked in by vboxsync, 14 years ago

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