VirtualBox

source: vbox/trunk/src/VBox/Runtime/r3/posix/pipe-posix.cpp@ 32625

Last change on this file since 32625 was 29559, checked in by vboxsync, 15 years ago

pipe-posix.cpp: nits
  • Property svn:eol-style set to native
  • Property svn:keywords set to Author Date Id Revision
File size: 16.5 KB
Line 
1/* $Id: pipe-posix.cpp 29559 2010-05-17 15:07:02Z vboxsync $ */
2/** @file
3 * IPRT - Anonymous Pipes, POSIX Implementation.
4 */
5
6/*
7 * Copyright (C) 2010 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#include <iprt/pipe.h>
32#include "internal/iprt.h"
33
34#include <iprt/asm.h>
35#include <iprt/assert.h>
36#include <iprt/err.h>
37#include <iprt/mem.h>
38#include <iprt/string.h>
39#include <iprt/thread.h>
40#include "internal/magics.h"
41
42#include <errno.h>
43#include <fcntl.h>
44#include <limits.h>
45#include <unistd.h>
46#include <sys/poll.h>
47#include <sys/stat.h>
48#include <signal.h>
49
50
51/*******************************************************************************
52* Structures and Typedefs *
53*******************************************************************************/
54typedef struct RTPIPEINTERNAL
55{
56 /** Magic value (RTPIPE_MAGIC). */
57 uint32_t u32Magic;
58 /** The file descriptor. */
59 int fd;
60 /** Set if this is the read end, clear if it's the write end. */
61 bool fRead;
62 /** Atomically operated state variable.
63 *
64 * - Bits 0 thru 29 - Users of the new mode.
65 * - Bit 30 - The pipe mode, set indicates blocking.
66 * - Bit 31 - Set when we're switching the mode.
67 */
68 uint32_t volatile u32State;
69} RTPIPEINTERNAL;
70
71
72/*******************************************************************************
73* Defined Constants And Macros *
74*******************************************************************************/
75/** @name RTPIPEINTERNAL::u32State defines
76 * @{ */
77#define RTPIPE_POSIX_BLOCKING UINT32_C(0x40000000)
78#define RTPIPE_POSIX_SWITCHING UINT32_C(0x80000000)
79#define RTPIPE_POSIX_SWITCHING_BIT 31
80#define RTPIPE_POSIX_USERS_MASK UINT32_C(0x3fffffff)
81/** @} */
82
83
84RTDECL(int) RTPipeCreate(PRTPIPE phPipeRead, PRTPIPE phPipeWrite, uint32_t fFlags)
85{
86 AssertPtrReturn(phPipeRead, VERR_INVALID_POINTER);
87 AssertPtrReturn(phPipeWrite, VERR_INVALID_POINTER);
88 AssertReturn(!(fFlags & ~RTPIPE_C_VALID_MASK), VERR_INVALID_PARAMETER);
89
90 /*
91 * Create the pipe and set the close-on-exec flag if requested.
92 */
93 int aFds[2] = {-1, -1};
94 if (pipe(aFds))
95 return RTErrConvertFromErrno(errno);
96
97 int rc = VINF_SUCCESS;
98 if (!(fFlags & RTPIPE_C_INHERIT_READ))
99 {
100 if (fcntl(aFds[0], F_SETFD, FD_CLOEXEC))
101 rc = RTErrConvertFromErrno(errno);
102 }
103
104 if (!(fFlags & RTPIPE_C_INHERIT_WRITE))
105 {
106 if (fcntl(aFds[1], F_SETFD, FD_CLOEXEC))
107 rc = RTErrConvertFromErrno(errno);
108 }
109
110 if (RT_SUCCESS(rc))
111 {
112 /*
113 * Create the two handles.
114 */
115 RTPIPEINTERNAL *pThisR = (RTPIPEINTERNAL *)RTMemAlloc(sizeof(RTPIPEINTERNAL));
116 if (pThisR)
117 {
118 RTPIPEINTERNAL *pThisW = (RTPIPEINTERNAL *)RTMemAlloc(sizeof(RTPIPEINTERNAL));
119 if (pThisW)
120 {
121 pThisR->u32Magic = RTPIPE_MAGIC;
122 pThisW->u32Magic = RTPIPE_MAGIC;
123 pThisR->fd = aFds[0];
124 pThisW->fd = aFds[1];
125 pThisR->fRead = true;
126 pThisW->fRead = false;
127 pThisR->u32State = RTPIPE_POSIX_BLOCKING;
128 pThisW->u32State = RTPIPE_POSIX_BLOCKING;
129
130 *phPipeRead = pThisR;
131 *phPipeWrite = pThisW;
132
133 /*
134 * Before we leave, make sure to shut up SIGPIPE.
135 */
136 signal(SIGPIPE, SIG_IGN);
137 return VINF_SUCCESS;
138 }
139
140 RTMemFree(pThisR);
141 rc = VERR_NO_MEMORY;
142 }
143 else
144 rc = VERR_NO_MEMORY;
145 }
146
147 close(aFds[0]);
148 close(aFds[1]);
149 return rc;
150}
151
152
153RTDECL(int) RTPipeClose(RTPIPE hPipe)
154{
155 RTPIPEINTERNAL *pThis = hPipe;
156 if (pThis == NIL_RTPIPE)
157 return VINF_SUCCESS;
158 AssertPtrReturn(pThis, VERR_INVALID_PARAMETER);
159 AssertReturn(pThis->u32Magic == RTPIPE_MAGIC, VERR_INVALID_HANDLE);
160
161 /*
162 * Do the cleanup.
163 */
164 AssertReturn(ASMAtomicCmpXchgU32(&pThis->u32Magic, ~RTPIPE_MAGIC, RTPIPE_MAGIC), VERR_INVALID_HANDLE);
165
166 int fd = pThis->fd;
167 pThis->fd = -1;
168 close(fd);
169
170 if (ASMAtomicReadU32(&pThis->u32State) & RTPIPE_POSIX_USERS_MASK)
171 {
172 AssertFailed();
173 RTThreadSleep(1);
174 }
175
176 RTMemFree(pThis);
177
178 return VINF_SUCCESS;
179}
180
181RTDECL(int) RTPipeFromNative(PRTPIPE phPipe, RTHCINTPTR hNativePipe, uint32_t fFlags)
182{
183 AssertPtrReturn(phPipe, VERR_INVALID_POINTER);
184 AssertReturn(!(fFlags & ~RTPIPE_N_VALID_MASK), VERR_INVALID_PARAMETER);
185 AssertReturn(!!(fFlags & RTPIPE_N_READ) != !!(fFlags & RTPIPE_N_WRITE), VERR_INVALID_PARAMETER);
186
187 /*
188 * Get and validate the pipe handle info.
189 */
190 int hNative = (int)hNativePipe;
191 struct stat st;
192 AssertReturn(fstat(hNative, &st) == 0, RTErrConvertFromErrno(errno));
193 AssertMsgReturn(S_ISFIFO(st.st_mode) || S_ISSOCK(st.st_mode), ("%#x (%o)\n", st.st_mode, st.st_mode), VERR_INVALID_HANDLE);
194
195 int fFd = fcntl(hNative, F_GETFL, 0);
196 AssertReturn(fFd != -1, VERR_INVALID_HANDLE);
197 AssertMsgReturn( (fFd & O_ACCMODE) == (fFlags & RTPIPE_N_READ ? O_RDONLY : O_WRONLY)
198 || (fFd & O_ACCMODE) == O_RDWR /* Solaris creates bi-directional pipes. */
199 , ("%#x\n", fFd), VERR_INVALID_HANDLE);
200
201 /*
202 * Create the handle.
203 */
204 RTPIPEINTERNAL *pThis = (RTPIPEINTERNAL *)RTMemAlloc(sizeof(RTPIPEINTERNAL));
205 if (!pThis)
206 return VERR_NO_MEMORY;
207
208 pThis->u32Magic = RTPIPE_MAGIC;
209 pThis->fd = hNative;
210 pThis->fRead = !!(fFlags & RTPIPE_N_READ);
211 pThis->u32State = fFd & O_NONBLOCK ? 0 : RTPIPE_POSIX_BLOCKING;
212
213 /*
214 * Fix up inheritability and shut up SIGPIPE and we're done.
215 */
216 if (fcntl(hNative, F_SETFD, fFlags & RTPIPE_N_INHERIT ? 0 : FD_CLOEXEC) == 0)
217 {
218 signal(SIGPIPE, SIG_IGN);
219 *phPipe = pThis;
220 return VINF_SUCCESS;
221 }
222
223 int rc = RTErrConvertFromErrno(errno);
224 RTMemFree(pThis);
225 return rc;
226}
227
228
229RTDECL(RTHCINTPTR) RTPipeToNative(RTPIPE hPipe)
230{
231 RTPIPEINTERNAL *pThis = hPipe;
232 AssertPtrReturn(pThis, -1);
233 AssertReturn(pThis->u32Magic == RTPIPE_MAGIC, -1);
234
235 return pThis->fd;
236}
237
238
239/**
240 * Prepare blocking mode.
241 *
242 * @returns VINF_SUCCESS
243 * @retval VERR_WRONG_ORDER
244 * @retval VERR_INTERNAL_ERROR_4
245 *
246 * @param pThis The pipe handle.
247 */
248static int rtPipeTryBlocking(RTPIPEINTERNAL *pThis)
249{
250 /*
251 * Update the state.
252 */
253 for (;;)
254 {
255 uint32_t u32State = ASMAtomicReadU32(&pThis->u32State);
256 uint32_t const u32StateOld = u32State;
257 uint32_t const cUsers = (u32State & RTPIPE_POSIX_USERS_MASK);
258
259 if (u32State & RTPIPE_POSIX_BLOCKING)
260 {
261 AssertReturn(cUsers < RTPIPE_POSIX_USERS_MASK / 2, VERR_INTERNAL_ERROR_4);
262 u32State &= ~RTPIPE_POSIX_USERS_MASK;
263 u32State |= cUsers + 1;
264 if (ASMAtomicCmpXchgU32(&pThis->u32State, u32State, u32StateOld))
265 {
266 if (u32State & RTPIPE_POSIX_SWITCHING)
267 break;
268 return VINF_SUCCESS;
269 }
270 }
271 else if (cUsers == 0)
272 {
273 u32State = 1 | RTPIPE_POSIX_SWITCHING | RTPIPE_POSIX_BLOCKING;
274 if (ASMAtomicCmpXchgU32(&pThis->u32State, u32State, u32StateOld))
275 break;
276 }
277 else
278 return VERR_WRONG_ORDER;
279 ASMNopPause();
280 }
281
282 /*
283 * Do the switching.
284 */
285 int fFlags = fcntl(pThis->fd, F_GETFL, 0);
286 if (fFlags != -1)
287 {
288 if ( !(fFlags & O_NONBLOCK)
289 || fcntl(pThis->fd, F_SETFL, fFlags & ~O_NONBLOCK) != -1)
290 {
291 ASMAtomicBitClear(&pThis->u32State, RTPIPE_POSIX_SWITCHING_BIT);
292 return VINF_SUCCESS;
293 }
294 }
295
296 ASMAtomicDecU32(&pThis->u32State);
297 return RTErrConvertFromErrno(errno);
298}
299
300
301/**
302 * Prepare non-blocking mode.
303 *
304 * @returns VINF_SUCCESS
305 * @retval VERR_WRONG_ORDER
306 * @retval VERR_INTERNAL_ERROR_4
307 *
308 * @param pThis The pipe handle.
309 */
310static int rtPipeTryNonBlocking(RTPIPEINTERNAL *pThis)
311{
312 /*
313 * Update the state.
314 */
315 for (;;)
316 {
317 uint32_t u32State = ASMAtomicReadU32(&pThis->u32State);
318 uint32_t const u32StateOld = u32State;
319 uint32_t const cUsers = (u32State & RTPIPE_POSIX_USERS_MASK);
320
321 if (!(u32State & RTPIPE_POSIX_BLOCKING))
322 {
323 AssertReturn(cUsers < RTPIPE_POSIX_USERS_MASK / 2, VERR_INTERNAL_ERROR_4);
324 u32State &= ~RTPIPE_POSIX_USERS_MASK;
325 u32State |= cUsers + 1;
326 if (ASMAtomicCmpXchgU32(&pThis->u32State, u32State, u32StateOld))
327 {
328 if (u32State & RTPIPE_POSIX_SWITCHING)
329 break;
330 return VINF_SUCCESS;
331 }
332 }
333 else if (cUsers == 0)
334 {
335 u32State = 1 | RTPIPE_POSIX_SWITCHING;
336 if (ASMAtomicCmpXchgU32(&pThis->u32State, u32State, u32StateOld))
337 break;
338 }
339 else
340 return VERR_WRONG_ORDER;
341 ASMNopPause();
342 }
343
344 /*
345 * Do the switching.
346 */
347 int fFlags = fcntl(pThis->fd, F_GETFL, 0);
348 if (fFlags != -1)
349 {
350 if ( (fFlags & O_NONBLOCK)
351 || fcntl(pThis->fd, F_SETFL, fFlags | O_NONBLOCK) != -1)
352 {
353 ASMAtomicBitClear(&pThis->u32State, RTPIPE_POSIX_SWITCHING_BIT);
354 return VINF_SUCCESS;
355 }
356 }
357
358 ASMAtomicDecU32(&pThis->u32State);
359 return RTErrConvertFromErrno(errno);
360}
361
362
363/**
364 * Checks if the read pipe has a HUP condition.
365 *
366 * @returns true if HUP, false if no.
367 * @param pThis The pipe handle (read).
368 */
369static bool rtPipePosixHasHup(RTPIPEINTERNAL *pThis)
370{
371 Assert(pThis->fRead);
372
373 struct pollfd PollFd;
374 RT_ZERO(PollFd);
375 PollFd.fd = pThis->fd;
376 PollFd.events = POLLHUP;
377 return poll(&PollFd, 1, 0) >= 1
378 && (PollFd.revents & POLLHUP);
379}
380
381
382RTDECL(int) RTPipeRead(RTPIPE hPipe, void *pvBuf, size_t cbToRead, size_t *pcbRead)
383{
384 RTPIPEINTERNAL *pThis = hPipe;
385 AssertPtrReturn(pThis, VERR_INVALID_HANDLE);
386 AssertReturn(pThis->u32Magic == RTPIPE_MAGIC, VERR_INVALID_HANDLE);
387 AssertReturn(pThis->fRead, VERR_ACCESS_DENIED);
388 AssertPtr(pcbRead);
389 AssertPtr(pvBuf);
390
391 int rc = rtPipeTryNonBlocking(pThis);
392 if (RT_SUCCESS(rc))
393 {
394 ssize_t cbRead = read(pThis->fd, pvBuf, RT_MIN(cbToRead, SSIZE_MAX));
395 if (cbRead >= 0)
396 {
397 if (cbRead || !cbToRead || !rtPipePosixHasHup(pThis))
398 *pcbRead = cbRead;
399 else
400 rc = VERR_BROKEN_PIPE;
401 }
402 else if (errno == EAGAIN)
403 {
404 *pcbRead = 0;
405 rc = VINF_TRY_AGAIN;
406 }
407 else
408 rc = RTErrConvertFromErrno(errno);
409
410 ASMAtomicDecU32(&pThis->u32State);
411 }
412 return rc;
413}
414
415
416RTDECL(int) RTPipeReadBlocking(RTPIPE hPipe, void *pvBuf, size_t cbToRead, size_t *pcbRead)
417{
418 RTPIPEINTERNAL *pThis = hPipe;
419 AssertPtrReturn(pThis, VERR_INVALID_HANDLE);
420 AssertReturn(pThis->u32Magic == RTPIPE_MAGIC, VERR_INVALID_HANDLE);
421 AssertReturn(pThis->fRead, VERR_ACCESS_DENIED);
422 AssertPtr(pvBuf);
423
424 int rc = rtPipeTryBlocking(pThis);
425 if (RT_SUCCESS(rc))
426 {
427 size_t cbTotalRead = 0;
428 while (cbToRead > 0)
429 {
430 ssize_t cbRead = read(pThis->fd, pvBuf, RT_MIN(cbToRead, SSIZE_MAX));
431 if (cbRead < 0)
432 {
433 rc = RTErrConvertFromErrno(errno);
434 break;
435 }
436 if (!cbRead && rtPipePosixHasHup(pThis))
437 {
438 rc = VERR_BROKEN_PIPE;
439 break;
440 }
441
442 /* advance */
443 pvBuf = (char *)pvBuf + cbRead;
444 cbTotalRead += cbRead;
445 cbToRead -= cbRead;
446 }
447
448 if (pcbRead)
449 {
450 *pcbRead = cbTotalRead;
451 if ( RT_FAILURE(rc)
452 && cbTotalRead
453 && rc != VERR_INVALID_POINTER)
454 rc = VINF_SUCCESS;
455 }
456
457 ASMAtomicDecU32(&pThis->u32State);
458 }
459 return rc;
460}
461
462
463RTDECL(int) RTPipeWrite(RTPIPE hPipe, const void *pvBuf, size_t cbToWrite, size_t *pcbWritten)
464{
465 RTPIPEINTERNAL *pThis = hPipe;
466 AssertPtrReturn(pThis, VERR_INVALID_HANDLE);
467 AssertReturn(pThis->u32Magic == RTPIPE_MAGIC, VERR_INVALID_HANDLE);
468 AssertReturn(!pThis->fRead, VERR_ACCESS_DENIED);
469 AssertPtr(pcbWritten);
470 AssertPtr(pvBuf);
471
472 int rc = rtPipeTryNonBlocking(pThis);
473 if (RT_SUCCESS(rc))
474 {
475 if (cbToWrite)
476 {
477 ssize_t cbWritten = write(pThis->fd, pvBuf, RT_MIN(cbToWrite, SSIZE_MAX));
478 if (cbWritten >= 0)
479 *pcbWritten = cbWritten;
480 else if (errno == EAGAIN)
481 {
482 *pcbWritten = 0;
483 rc = VINF_TRY_AGAIN;
484 }
485 else
486 rc = RTErrConvertFromErrno(errno);
487 }
488 else
489 *pcbWritten = 0;
490
491 ASMAtomicDecU32(&pThis->u32State);
492 }
493 return rc;
494}
495
496
497RTDECL(int) RTPipeWriteBlocking(RTPIPE hPipe, const void *pvBuf, size_t cbToWrite, size_t *pcbWritten)
498{
499 RTPIPEINTERNAL *pThis = hPipe;
500 AssertPtrReturn(pThis, VERR_INVALID_HANDLE);
501 AssertReturn(pThis->u32Magic == RTPIPE_MAGIC, VERR_INVALID_HANDLE);
502 AssertReturn(!pThis->fRead, VERR_ACCESS_DENIED);
503 AssertPtr(pvBuf);
504 AssertPtrNull(pcbWritten);
505
506 int rc = rtPipeTryBlocking(pThis);
507 if (RT_SUCCESS(rc))
508 {
509 size_t cbTotalWritten = 0;
510 while (cbToWrite > 0)
511 {
512 ssize_t cbWritten = write(pThis->fd, pvBuf, RT_MIN(cbToWrite, SSIZE_MAX));
513 if (cbWritten < 0)
514 {
515 rc = RTErrConvertFromErrno(errno);
516 break;
517 }
518
519 /* advance */
520 pvBuf = (char const *)pvBuf + cbWritten;
521 cbTotalWritten += cbWritten;
522 cbToWrite -= cbWritten;
523 }
524
525 if (pcbWritten)
526 {
527 *pcbWritten = cbTotalWritten;
528 if ( RT_FAILURE(rc)
529 && cbTotalWritten
530 && rc != VERR_INVALID_POINTER)
531 rc = VINF_SUCCESS;
532 }
533
534 ASMAtomicDecU32(&pThis->u32State);
535 }
536 return rc;
537}
538
539
540RTDECL(int) RTPipeFlush(RTPIPE hPipe)
541{
542 RTPIPEINTERNAL *pThis = hPipe;
543 AssertPtrReturn(pThis, VERR_INVALID_HANDLE);
544 AssertReturn(pThis->u32Magic == RTPIPE_MAGIC, VERR_INVALID_HANDLE);
545 AssertReturn(!pThis->fRead, VERR_ACCESS_DENIED);
546
547 if (fsync(pThis->fd))
548 {
549 if (errno == EINVAL || errno == ENOTSUP)
550 return VERR_NOT_SUPPORTED;
551 return RTErrConvertFromErrno(errno);
552 }
553 return VINF_SUCCESS;
554}
555
556
557RTDECL(int) RTPipeSelectOne(RTPIPE hPipe, RTMSINTERVAL cMillies)
558{
559 RTPIPEINTERNAL *pThis = hPipe;
560 AssertPtrReturn(pThis, VERR_INVALID_HANDLE);
561 AssertReturn(pThis->u32Magic == RTPIPE_MAGIC, VERR_INVALID_HANDLE);
562
563 struct pollfd PollFd;
564 RT_ZERO(PollFd);
565 PollFd.fd = pThis->fd;
566 PollFd.events = POLLHUP | POLLERR;
567 if (pThis->fRead)
568 PollFd.events |= POLLIN | POLLPRI;
569 else
570 PollFd.events |= POLLOUT;
571
572 int timeout;
573 if ( cMillies == RT_INDEFINITE_WAIT
574 || cMillies >= INT_MAX /* lazy bird */)
575 timeout = -1;
576 else
577 timeout = cMillies;
578
579 int rc = poll(&PollFd, 1, 0);
580 if (rc == -1)
581 return RTErrConvertFromErrno(errno);
582 return rc > 0 ? VINF_SUCCESS : VERR_TIMEOUT;
583}
584
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