From 424116620a78ef29a21c2dfdce4c9902cb53b4df Mon Sep 17 00:00:00 2001 From: Edmond <1571649+edmonddantes@users.noreply.github.com> Date: Thu, 8 Oct 2026 15:23:37 +0000 Subject: [PATCH] Make proc_open() pipes overlapped named pipes on Windows when a provider asks An anonymous pipe takes no overlapped I/O, so a coroutine reading a proc_open() pipe blocked the thread, and stream_select() on pipes only polled. With php_io_overlapped_pipes set (by a provider at MINIT), the parent's end of each 'pipe' descriptor is a named pipe created with FILE_FLAG_OVERLAPPED, FILE_FLAG_FIRST_PIPE_INSTANCE and 64 KB buffers, and the child's end is a synchronous CreateFile() handle, as libuv's uv_spawn() makes them. The parent's end opens in binary mode, since its reads and writes go around the CRT. Without the setting nothing changes; STDIN and exec pipes stay as they are. Reads and writes of such a pipe (PHP_IO_OP_F_PIPE) go to the provider. Without one, or when it answers Unsupported, the thread waits on an overlapped request whose event has the low-order bit set, so no packet is queued; a write to a pipe whose reader closed fails with EPIPE. stream_set_blocking(false) is kept on the stream, since the handle has no non-blocking mode: a read then returns only the bytes the pipe holds. stream_select() asks the provider for read readiness with a zero-byte overlapped read; a pipe in the write set is always writable, and one in the except set is answered as php_select() answers it. When another process holds a passed pipe, a member is a descriptor no socket poll takes, or no provider can wait for a pipe, php_select() takes the call (for the time left). php_select() judges every pipe by the bytes it holds, since an overlapped pipe's handle is signaled whatever it holds. A cast fails as a concurrent access while another call has an op on the stream, and waits for an orphaned one. Right before CreateProcessW(), every queue of the thread lets go of each pipe passed to the child (queue op release(); the Ring calls ior_release_handle()), so the child's own overlapped I/O does not post to the parent's port. A pipe with an op in flight fails proc_open() as a concurrent access; where ior cannot release the handle (an op of its own still on it, a system before Windows 8.1), it drops the child's packets instead. After CreateProcessW() the pipe is marked handed out and the parent's later calls on it skip the provider. --- ext/standard/proc_open.c | 130 ++++++++++++++++++- ext/standard/streamsfuncs.c | 136 ++++++++++++++++++-- main/io/php_io_hooks.c | 159 +++++++++++++++++++++++- main/io/php_io_ring.c | 155 ++++++++++++++++++++++- main/php_io_hooks.h | 43 +++++++ main/php_streams.h | 11 ++ main/streams/php_stream_plain_wrapper.h | 6 + main/streams/plain_wrapper.c | 87 ++++++++++++- win32/ioutil.c | 7 ++ win32/ioutil.h | 1 + win32/select.c | 64 +++++++++- 11 files changed, 767 insertions(+), 32 deletions(-) diff --git a/ext/standard/proc_open.c b/ext/standard/proc_open.c index 772f7c080904..6962e800c3df 100644 --- a/ext/standard/proc_open.c +++ b/ext/standard/proc_open.c @@ -23,6 +23,7 @@ #include "zend_smart_str.h" #ifdef PHP_WIN32 # include "win32/sockets.h" +# include "ext/random/php_random_csprng.h" #endif #ifdef HAVE_SYS_WAIT_H @@ -913,18 +914,122 @@ static php_file_descriptor_t make_descriptor_cloexec(php_file_descriptor_t fd) #endif } +#ifdef PHP_WIN32 +/* Makes a child's pipe as libuv's uv_spawn() does: the parent's end is an overlapped named pipe, + * which an I/O completion port can complete, and the child's end is synchronous and inheritable. + * One instance and a name that cannot be guessed, so no other client can take the child's place. + * pair is filled as pipe() fills it: the read end, then the write end. */ +static zend_result create_overlapped_pipe(HANDLE pair[2], bool parent_reads) +{ + const DWORD parent_access = (parent_reads ? PIPE_ACCESS_INBOUND : PIPE_ACCESS_OUTBOUND) + | FILE_FLAG_OVERLAPPED | FILE_FLAG_FIRST_PIPE_INSTANCE | WRITE_DAC; + const DWORD child_access = (parent_reads ? GENERIC_WRITE | FILE_READ_ATTRIBUTES : GENERIC_READ | FILE_WRITE_ATTRIBUTES) + | WRITE_DAC; + wchar_t name[64]; + HANDLE server = INVALID_HANDLE_VALUE; + DWORD err = ERROR_SUCCESS; + + for (int attempt = 0; attempt < 16; attempt++) { + uint64_t name_bits; + if (php_random_bytes_silent(&name_bits, sizeof(name_bits)) == FAILURE) { + errno = EIO; + return FAILURE; + } + + _snwprintf_s(name, sizeof(name) / sizeof(name[0]), _TRUNCATE, L"\\\\.\\pipe\\php-%lu-%016llx", + GetCurrentProcessId(), (unsigned long long) name_bits); + server = CreateNamedPipeW(name, parent_access, + PIPE_TYPE_BYTE | PIPE_READMODE_BYTE | PIPE_WAIT | PIPE_REJECT_REMOTE_CLIENTS, 1, + PHP_IO_PIPE_BUFFER_SIZE, PHP_IO_PIPE_BUFFER_SIZE, 0, NULL); + if (server != INVALID_HANDLE_VALUE) { + break; + } + + err = GetLastError(); + /* A taken name: ERROR_ACCESS_DENIED under FILE_FLAG_FIRST_PIPE_INSTANCE, else ERROR_PIPE_BUSY */ + if (err != ERROR_ACCESS_DENIED && err != ERROR_PIPE_BUSY) { + break; + } + } + + if (server == INVALID_HANDLE_VALUE) { + SET_ERRNO_FROM_WIN32_CODE(err); + return FAILURE; + } + + const HANDLE client = CreateFileW(name, child_access, 0, &php_proc_open_security, OPEN_EXISTING, 0, NULL); + if (client == INVALID_HANDLE_VALUE) { + SET_ERRNO_FROM_WIN32_CODE(GetLastError()); + CloseHandle(server); + return FAILURE; + } + + /* Both ends exist, so the connect completes at once; the wait keeps a late completion off the + * stack */ + OVERLAPPED ov; + memset(&ov, 0, sizeof(ov)); + BOOL connected = ConnectNamedPipe(server, &ov); + if (!connected) { + const DWORD connect_err = GetLastError(); + DWORD unused; + connected = connect_err == ERROR_PIPE_CONNECTED + || (connect_err == ERROR_IO_PENDING && GetOverlappedResult(server, &ov, &unused, TRUE)); + } + + if (!connected) { + SET_ERRNO_FROM_WIN32_CODE(GetLastError()); + CloseHandle(client); + CloseHandle(server); + return FAILURE; + } + + pair[0] = parent_reads ? server : client; + pair[1] = parent_reads ? client : server; + return SUCCESS; +} + +/* PHP_STREAM_OPTION_OVERLAPPED_PIPE on every stream of the spec, up to the first that fails (only + * RELEASE can); one closed meanwhile is skipped */ +static zend_result set_passed_streams_pipe_option(const HashTable *descriptorspec, int value) +{ + zval *descitem; + + ZEND_HASH_FOREACH_VAL(descriptorspec, descitem) { + ZVAL_DEREF(descitem); + if (Z_TYPE_P(descitem) != IS_RESOURCE) { + continue; + } + + php_stream *const passed = (php_stream *) zend_fetch_resource(Z_RES_P(descitem), NULL, php_file_le_stream()); + if (passed && php_stream_set_option(passed, PHP_STREAM_OPTION_OVERLAPPED_PIPE, value, NULL) + == PHP_STREAM_OPTION_RETURN_ERR) { + return FAILURE; + } + } ZEND_HASH_FOREACH_END(); + + return SUCCESS; +} +#endif + static zend_result set_proc_descriptor_to_pipe(descriptorspec_item *desc, zend_string *zmode) { php_file_descriptor_t newpipe[2]; + const bool parent_reads = zend_string_starts_with_literal(zmode, "w"); - if (pipe(newpipe)) { +#ifdef PHP_WIN32 + const bool failed = php_io_overlapped_pipes ? create_overlapped_pipe(newpipe, parent_reads) == FAILURE + : pipe(newpipe) != 0; +#else + const bool failed = pipe(newpipe) != 0; +#endif + if (failed) { php_error_docref(NULL, E_WARNING, "Unable to create pipe %s", strerror(errno)); return FAILURE; } desc->type = DESCRIPTOR_TYPE_PIPE; - if (!zend_string_starts_with_literal(zmode, "w")) { + if (!parent_reads) { desc->parentend = newpipe[1]; desc->childend = newpipe[0]; desc->mode_flags = O_WRONLY; @@ -937,7 +1042,8 @@ static zend_result set_proc_descriptor_to_pipe(descriptorspec_item *desc, zend_s desc->parentend = make_descriptor_cloexec(desc->parentend); #ifdef PHP_WIN32 - if (ZSTR_LEN(zmode) >= 2 && ZSTR_VAL(zmode)[1] == 'b') + /* An overlapped pipe's reads and writes bypass the CRT's text mode */ + if ((ZSTR_LEN(zmode) >= 2 && ZSTR_VAL(zmode)[1] == 'b') || php_io_overlapped_pipes) desc->mode_flags |= O_BINARY; #endif @@ -1359,6 +1465,12 @@ PHP_FUNCTION(proc_open) goto exit_fail; } + /* Not in the descriptor loop above, which runs PHP code: another coroutine's read there would + * tie the pipe to a completion port again */ + if (set_passed_streams_pipe_option(descriptorspec, PHP_STREAM_OVERLAPPED_PIPE_RELEASE) == FAILURE) { + goto exit_fail; + } + init_startup_info(&si, descriptors, ndesc); init_process_info(&pi); @@ -1415,6 +1527,12 @@ PHP_FUNCTION(proc_open) childHandle = pi.hProcess; child = pi.dwProcessId; CloseHandle(pi.hThread); + + /* The child holds the descriptors of the streams passed to it: the parent's own calls on an + * overlapped pipe among them stop using the provider for good, since a child may bind the pipe + * to its own port with FILE_SKIP_COMPLETION_PORT_ON_SUCCESS, as libuv does, and a Ring op that + * completes at once would then never post. */ + set_passed_streams_pipe_option(descriptorspec, PHP_STREAM_OVERLAPPED_PIPE_HAND_OUT); #elif defined(USE_POSIX_SPAWN) posix_spawn_file_actions_t factions; int r; @@ -1554,8 +1672,10 @@ PHP_FUNCTION(proc_open) } #ifdef PHP_WIN32 - stream = php_stream_fopen_from_fd(_open_osfhandle((intptr_t)descriptors[i].parentend, - descriptors[i].mode_flags), mode_string, NULL); + const int parent_fd = _open_osfhandle((intptr_t)descriptors[i].parentend, descriptors[i].mode_flags); + stream = php_io_overlapped_pipes + ? php_stream_fopen_from_overlapped_pipe(parent_fd, mode_string) + : php_stream_fopen_from_fd(parent_fd, mode_string, NULL); php_stream_set_option(stream, PHP_STREAM_OPTION_PIPE_BLOCKING, blocking_pipes, NULL); #else stream = php_stream_fopen_from_fd(descriptors[i].parentend, mode_string, NULL); diff --git a/ext/standard/streamsfuncs.c b/ext/standard/streamsfuncs.c index e6a8222bb2b7..df84451e2870 100644 --- a/ext/standard/streamsfuncs.c +++ b/ext/standard/streamsfuncs.c @@ -691,9 +691,46 @@ typedef struct php_select_member { php_stream *stream; php_socket_t fd; uint32_t events; +#ifdef PHP_WIN32 + bool is_overlapped_pipe; /* PHP_IO_OP_F_PIPE: readable when it holds bytes, no socket poll takes it */ + bool in_except; /* in the except set: a backend without priority events drops it from events */ +#endif } php_select_member; -static void stream_array_collect_members(HashTable *stream_array, uint32_t events, +#ifdef PHP_WIN32 +/* stream_select_any(): a member no provider can wait for, so php_select() takes the call */ +# define STREAM_SELECT_USE_PHP_SELECT (-2) + +/* A pipe has no write readiness: always writable, and in the except set as php_select() answers */ +static void stream_select_report_pipe(const php_select_member *m, fd_set *rfds, fd_set *wfds, fd_set *efds, + int *found) +{ + if ((m->events & PHP_POLL_READ) && php_io_pipe_readable((int) m->fd)) { + PHP_SAFE_FD_SET(m->fd, rfds); + (*found)++; + } + + if (m->events & PHP_POLL_WRITE) { + PHP_SAFE_FD_SET(m->fd, wfds); + (*found)++; + } + + if (m->in_except) { + PHP_SAFE_FD_SET(m->fd, efds); + (*found)++; + } +} + +static bool stream_select_is_socket(php_socket_t fd) +{ + int type; + int len = sizeof(type); + return getsockopt((SOCKET) fd, SOL_SOCKET, SO_TYPE, (char *) &type, &len) == 0 + || WSAGetLastError() != WSAENOTSOCK; +} +#endif + +static void stream_array_collect_members(HashTable *stream_array, uint32_t events, bool except, php_select_member **members, uint32_t *n, uint32_t *cap) { zval *elem; @@ -713,6 +750,9 @@ static void stream_array_collect_members(HashTable *stream_array, uint32_t event for (uint32_t i = 0; i < *n; i++) { if ((*members)[i].fd == this_fd) { (*members)[i].events |= events; +#ifdef PHP_WIN32 + (*members)[i].in_except |= except; +#endif merged = true; break; } @@ -725,6 +765,9 @@ static void stream_array_collect_members(HashTable *stream_array, uint32_t event (*members)[*n].stream = stream; (*members)[*n].fd = this_fd; (*members)[*n].events = events; +#ifdef PHP_WIN32 + (*members)[*n].in_except = except; +#endif (*n)++; } } ZEND_HASH_FOREACH_END(); @@ -799,14 +842,34 @@ static int stream_select_any(zval *r_array, zval *w_array, zval *e_array, struct } if (r_array) { - stream_array_collect_members(Z_ARRVAL_P(r_array), PHP_POLL_READ, &members, &n, &cap); + stream_array_collect_members(Z_ARRVAL_P(r_array), PHP_POLL_READ, false, &members, &n, &cap); } if (w_array) { - stream_array_collect_members(Z_ARRVAL_P(w_array), PHP_POLL_WRITE, &members, &n, &cap); + stream_array_collect_members(Z_ARRVAL_P(w_array), PHP_POLL_WRITE, false, &members, &n, &cap); } if (e_array) { - stream_array_collect_members(Z_ARRVAL_P(e_array), priority ? PHP_POLL_PRI : 0, &members, &n, &cap); + stream_array_collect_members(Z_ARRVAL_P(e_array), priority ? PHP_POLL_PRI : 0, true, &members, &n, &cap); + } + +#ifdef PHP_WIN32 + for (uint32_t i = 0; i < n; i++) { + bool handed_out = false; + const bool is_overlapped_pipe = php_stream_set_option(members[i].stream, + PHP_STREAM_OPTION_OVERLAPPED_PIPE, PHP_STREAM_OVERLAPPED_PIPE_QUERY, &handed_out) + == PHP_STREAM_OPTION_RETURN_OK; + /* php_select() takes the call when another process holds the pipe, or when no socket poll + * takes the descriptor (an anonymous pipe, a file, a console): WSAPoll() fails as a whole on + * one */ + if (handed_out || (php_io_overlapped_pipes && !is_overlapped_pipe + && !stream_select_is_socket(members[i].fd))) { + efree(members); + stream_select_unfreeze(frozen, n_frozen); + return STREAM_SELECT_USE_PHP_SELECT; + } + + members[i].is_overlapped_pipe = is_overlapped_pipe; } +#endif /* Syscall first, as for a read: a zero-timeout poll() over the sets answers a select * with a stream ready now, or one with a zero timeout, without an op, and is the arm-time @@ -817,18 +880,34 @@ static int stream_select_any(zval *r_array, zval *w_array, zval *e_array, struct if (n > 0) { php_pollfd stack[16]; php_pollfd *fds = n > sizeof(stack) / sizeof(stack[0]) ? safe_emalloc(n, sizeof(*fds), 0) : stack; + uint32_t n_fds = 0; + int found = 0; for (uint32_t i = 0; i < n; i++) { - fds[i].fd = members[i].fd; - fds[i].events = ((members[i].events & PHP_POLL_READ) ? POLLIN : 0) +#ifdef PHP_WIN32 + if (members[i].is_overlapped_pipe) { + continue; + } +#endif + + fds[n_fds].fd = members[i].fd; + fds[n_fds].events = ((members[i].events & PHP_POLL_READ) ? POLLIN : 0) | ((members[i].events & PHP_POLL_WRITE) ? POLLOUT : 0) | ((members[i].events & PHP_POLL_PRI) ? POLLPRI : 0); - fds[i].revents = 0; + fds[n_fds].revents = 0; + n_fds++; } - int ready = php_poll2(fds, n, 0); - int found = 0; + const int ready = n_fds > 0 ? php_poll2(fds, n_fds, 0) : 0; if (ready >= 0) { - for (uint32_t i = 0; i < n; i++) { - short revents = ready > 0 ? fds[i].revents : 0; + for (uint32_t i = 0, polled = 0; i < n; i++) { +#ifdef PHP_WIN32 + if (members[i].is_overlapped_pipe) { + stream_select_report_pipe(&members[i], rfds, wfds, efds, &found); + continue; + } +#endif + + const short revents = ready > 0 ? fds[polled].revents : 0; + polled++; if (revents & POLLNVAL) { /* The Any reports the descriptor's failure */ found = 0; @@ -879,6 +958,11 @@ static int stream_select_any(zval *r_array, zval *w_array, zval *e_array, struct if (checked) { ops[n_members].flags |= PHP_IO_OP_F_CHECKED; } +#ifdef PHP_WIN32 + if (members[i].is_overlapped_pipe) { + ops[n_members].flags |= PHP_IO_OP_F_PIPE; + } +#endif op_ptrs[n_members] = &ops[n_members]; n_members++; } @@ -906,6 +990,22 @@ static int stream_select_any(zval *r_array, zval *w_array, zval *e_array, struct errno = EINTR; ret = -1; } else { +#ifdef PHP_WIN32 + for (uint32_t i = 0; i < any.u.any.n_results; i++) { + if (results[i].status == PHP_IO_UNSUPPORTED && (ops[results[i].index].flags & PHP_IO_OP_F_PIPE)) { + /* No provider waits for this pipe: the thread does, for the time left */ + if (tv) { + const zend_hrtime_t left_ns = php_io_deadline_remaining(&ops[timer_index].deadline, + zend_hrtime()); + tv->tv_sec = (long) (left_ns / ZEND_NANO_IN_SEC); + tv->tv_usec = (long) (left_ns % ZEND_NANO_IN_SEC / 1000); + } + + ret = STREAM_SELECT_USE_PHP_SELECT; + goto done; + } + } +#endif for (uint32_t i = 0; i < any.u.any.n_results; i++) { uint32_t index = results[i].index; if (index == timer_index || (results[i].status != PHP_IO_DONE && results[i].status != PHP_IO_READY)) { @@ -930,6 +1030,9 @@ static int stream_select_any(zval *r_array, zval *w_array, zval *e_array, struct } } +#ifdef PHP_WIN32 +done: +#endif efree(results); efree(op_ptrs); efree(ops); @@ -1141,11 +1244,22 @@ PHP_FUNCTION(stream_select) } if (php_io_hooks_active()) { +#ifdef PHP_WIN32 + const fd_set all_rfds = rfds, all_wfds = wfds, all_efds = efds; +#endif /* The provider waits: the sets are rebuilt from what it reported */ FD_ZERO(&rfds); FD_ZERO(&wfds); FD_ZERO(&efds); retval = stream_select_any(r_array, w_array, e_array, tv_p, &rfds, &wfds, &efds); +#ifdef PHP_WIN32 + if (retval == STREAM_SELECT_USE_PHP_SELECT) { + rfds = all_rfds; + wfds = all_wfds; + efds = all_efds; + retval = php_select(max_fd+1, &rfds, &wfds, &efds, tv_p); + } +#endif } else { retval = php_select(max_fd+1, &rfds, &wfds, &efds, tv_p); } diff --git a/main/io/php_io_hooks.c b/main/io/php_io_hooks.c index 2a3a47201735..d5bd01d9de2b 100644 --- a/main/io/php_io_hooks.c +++ b/main/io/php_io_hooks.c @@ -34,6 +34,9 @@ extern int fdatasync(int); PHPAPI void (*php_io_op_zobj_detach)(zend_object *zobj) = NULL; PHPAPI bool (*php_io_signal_pending)(void) = NULL; +#ifdef PHP_WIN32 +PHPAPI bool php_io_overlapped_pipes = false; +#endif PHPAPI bool php_io_interrupt_pending(void) { @@ -434,6 +437,15 @@ PHPAPI php_io_queue *php_io_queue_find(uint64_t id) return NULL; } +PHPAPI void php_io_queues_release(php_socket_t fd) +{ + for (php_io_queue *q = php_io_queues; q; q = q->next) { + if (q->ops->release) { + q->ops->release(q, fd); + } + } +} + /* Only the pairs of the current provider: one an earlier provider registered may sit on a queue * nobody serves any more */ PHPAPI uint32_t php_io_held_events(php_io_registration *regs, uint32_t events) @@ -679,20 +691,32 @@ PHPAPI bool php_io_stream_busy(php_stream *stream) return (stream->flags & PHP_STREAM_FLAG_IN_USE) && !php_io_orphan_find(stream); } -/* Called from php_stream_free() with the stream still frozen */ -PHPAPI void php_io_stream_drain(php_stream *stream) +static void php_io_stream_wait_orphan(php_stream *stream, bool freeing) { php_io_orphan *o = php_io_orphan_find(stream); if (!o) { return; } - o->freeing = true; + o->freeing = freeing; if (o->queue->ops->drain) { o->queue->ops->drain(o->queue, stream); } php_io_stream_unfreeze(stream); } +/* Called from php_stream_free() with the stream still frozen */ +PHPAPI void php_io_stream_drain(php_stream *stream) +{ + php_io_stream_wait_orphan(stream, true); +} + +#ifdef PHP_WIN32 +PHPAPI void php_io_stream_settle(php_stream *stream) +{ + php_io_stream_wait_orphan(stream, false); +} +#endif + /* The op of a handle without a stream: its descriptor must stay open until the op settled */ PHPAPI void php_io_handle_orphan(zend_object *handle, php_io_queue *queue) { @@ -1783,6 +1807,135 @@ PHPAPI ssize_t php_io_write(php_stream *stream, int fd, const void *buf, size_t return php_io_write_at(stream, fd, buf, len, -1, dl); } +#ifdef PHP_WIN32 +/* Any offset >= 0 picks the overlapped call an overlapped handle needs; a pipe ignores the offset */ +#define PHP_IO_PIPE_OFFSET 0 + +/* Bytes the pipe holds now; *eof at the other end's close */ +static bool php_io_pipe_peek(int fd, DWORD *avail, bool *eof) +{ + *avail = 0; + *eof = false; + const HANDLE h = (HANDLE) _get_osfhandle(fd); + if (h == INVALID_HANDLE_VALUE) { + php_io_set_errno(EBADF); + return false; + } + + if (PeekNamedPipe(h, NULL, 0, NULL, avail, NULL)) { + return true; + } + + const DWORD err = GetLastError(); + if (err == ERROR_BROKEN_PIPE) { + *eof = true; + return true; + } + + /* ERROR_ACCESS_DENIED: fd is the write end */ + php_io_set_errno(err == ERROR_ACCESS_DENIED ? EBADF : EIO); + return false; +} + +PHPAPI bool php_io_pipe_readable(int fd) +{ + DWORD avail; + bool eof; + return !php_io_pipe_peek(fd, &avail, &eof) || eof || avail > 0; +} + +PHPAPI ssize_t php_io_pipe_read(php_stream *stream, int fd, void *buf, size_t len, + const php_deadline *dl, bool no_provider) +{ + if (php_io_stream_read_lost(stream)) { + return -1; + } + + php_io_frame f; + php_io_op op; + php_io_op_result result; + ssize_t ret; + + if (php_io_frame_begin(&f, stream) == FAILURE) { + return -1; + } + + memset(&op, 0, sizeof(op)); + + DWORD avail; + bool eof; + if (!php_io_pipe_peek(fd, &avail, &eof)) { + ret = -1; + } else if (eof) { + ret = 0; + } else if (avail > 0) { + /* Never more than is there, so the call does not wait */ + ret = php_io_read_syscall(fd, buf, MIN(len, avail), PHP_IO_PIPE_OFFSET); + } else if (dl->hrtime == 0) { + php_io_set_errno(EAGAIN); + ret = -1; + } else if (no_provider || !FG(io_hooks)) { + ret = php_io_read_syscall(fd, buf, len, PHP_IO_PIPE_OFFSET); + } else { + /* Offset -1: the bytes a cancelled read took are the stream's (php_io_op_read_advances) */ + php_io_op_read(&op, NULL, fd, buf, len, -1, php_io_deadline_infinite()); + op.flags |= PHP_IO_OP_F_PIPE | php_io_stream_buf_flag(stream, buf); + op.stream = stream; + if (php_io_run(&op, &result) == FAILURE || php_io_run_cancelled(stream, &op, &result)) { + php_io_set_errno(ECANCELED); + ret = -1; + } else if (php_io_data_result(&result, &ret)) { + if (ret < 0 && errno == EPIPE) { + /* The writer closed while the read waited */ + ret = 0; + } + } else { + /* Unsupported, or a Ready that did not wait: the thread waits for data */ + ret = php_io_read_syscall(fd, buf, len, PHP_IO_PIPE_OFFSET); + } + } + + php_io_frame_end(&f, &op); + return ret; +} + +PHPAPI ssize_t php_io_pipe_write(php_stream *stream, int fd, const void *buf, size_t len, + bool no_provider) +{ + php_io_frame f; + php_io_op op; + php_io_op_result result; + ssize_t ret; + + if (php_io_frame_begin(&f, stream) == FAILURE) { + return -1; + } + + memset(&op, 0, sizeof(op)); + + if (no_provider || !FG(io_hooks)) { + ret = php_io_write_syscall(fd, (void *) buf, len, PHP_IO_PIPE_OFFSET); + } else { + const size_t chunk = MIN(len, PHP_IO_PIPE_BUFFER_SIZE); + /* Infinite whatever the stream's mode: a pipe write cannot be partial, so it parks until done */ + php_io_op_write(&op, NULL, fd, buf, chunk, -1, php_io_deadline_infinite()); + op.flags |= PHP_IO_OP_F_PIPE; + op.stream = stream; + if (php_io_run(&op, &result) == FAILURE || php_io_run_cancelled(stream, &op, &result)) { + php_io_set_errno(ECANCELED); + ret = -1; + } else if (!php_io_data_result(&result, &ret)) { + /* Ready leaves the chunk to the thread, Unsupported the whole count */ + ret = php_io_write_syscall(fd, (void *) buf, result.status == PHP_IO_READY ? chunk : len, + PHP_IO_PIPE_OFFSET); + } + } + + php_io_frame_end(&f, &op); + return ret; +} +#endif + PHPAPI int php_io_fsync(php_stream *stream, int fd, bool data_only) { if (FG(io_hooks) && (php_io_hook_flags() & PHP_IO_HOOKS_F_FILES)) { diff --git a/main/io/php_io_ring.c b/main/io/php_io_ring.c index c6a85b86d2f3..c4551977bfbf 100644 --- a/main/io/php_io_ring.c +++ b/main/io/php_io_ring.c @@ -90,6 +90,11 @@ struct _php_io_ring_req { bool multishot; /* an Edge record's multishot poll, without an op */ bool sentinel; /* the entry that fires the heap's head for the notification descriptor */ php_io_ring_reg *reg; /* multishot: its record, NULL once removed */ +#ifdef PHP_WIN32 + uint32_t pipe_events; /* POLL with PHP_IO_OP_F_PIPE: the PHP_POLL_* it waits for, else 0 */ + char zero_read_buf; /* a pipe READ poll's zero-byte read: ReadFile() wants a buffer */ + php_stream *zero_read_stream; /* the stream whose drain waits for that read once it is cancelled */ +#endif php_io_ring_reg *waiting; /* a wait parked on an Edge record, without an entry */ uint32_t w_mask; /* IOR_POLL_* the parked wait wants */ php_io_ring_req *w_prev; /* the parked waits */ @@ -641,6 +646,11 @@ static php_io_ring_req *php_io_ring_req_create(php_io_ring *ring, php_io_op *op, req->data = data; req->type = op->type; req->deadline = op->deadline; +#ifdef PHP_WIN32 + if (op->type == PHP_IO_OP_POLL && (op->flags & PHP_IO_OP_F_PIPE)) { + req->pipe_events = op->u.poll.events & (PHP_POLL_READ | PHP_POLL_WRITE); + } +#endif php_io_ring_req_capture(req, op); return req; } @@ -985,6 +995,11 @@ static zend_result php_io_ring_submit_one(php_io_ring *ring, php_io_ring_req *re /* What has no ring form is refused before an entry is taken */ switch (req->type) { case PHP_IO_OP_POLL: +#ifdef PHP_WIN32 + if (req->pipe_events) { + break; + } +#endif /* ior has no priority event */ if (!(ring->features & IOR_FEAT_POLL_ADD) || (req->op->u.poll.events & PHP_POLL_PRI)) { errno = ENOTSUP; @@ -1016,6 +1031,20 @@ static zend_result php_io_ring_submit_one(php_io_ring *ring, php_io_ring_req *re switch (req->type) { case PHP_IO_OP_POLL: +#ifdef PHP_WIN32 + if (req->pipe_events & PHP_POLL_WRITE) { + /* A pipe has no write readiness: ready at once */ + ior_prep_nop(ctx, sqe); + break; + } + + if (req->pipe_events) { + /* A zero-byte read completes when data arrives and takes none of it, as libuv's zero + * reads do */ + ior_prep_read(ctx, sqe, php_io_ring_file_fd(op), &req->zero_read_buf, 0, IOR_OFF_NONE); + break; + } +#endif ior_prep_poll_add(ctx, sqe, (ior_fd_t) op->fd, php_io_ring_poll_mask_to_ior(op->u.poll.events)); break; case PHP_IO_OP_READ: @@ -1587,6 +1616,36 @@ static void php_io_ring_req_release(php_io_ring *ring, php_io_ring_req *req) } } +#ifdef PHP_WIN32 +/* A cancelled pipe READ poll lasts in the kernel until its cancel completes, and a synchronous read + * of the descriptor meanwhile, in this process or one it is handed to, may complete falsely: the + * member comes back in flight, and the provider's drain of the stream waits for it + * (PHP_IO_OP_F_PIPE) */ +static bool php_io_ring_zero_read_keep(php_io_ring_req *m) +{ + if ((m->pipe_events & PHP_POLL_READ) && m->op->stream && !php_io_ring_req_settled(m)) { + m->zero_read_stream = m->op->stream; + return true; + } + + return false; +} +#endif + +/* php_io_ring_req_release() of an Any's member: a pipe READ poll whose read is still in the kernel + * leaves its op in flight, and the waiter drains the stream */ +static void php_io_ring_member_release(php_io_ring *ring, php_io_ring_req *m) +{ +#ifdef PHP_WIN32 + php_io_op *const member_op = m->op; + const bool zero_read_kept = php_io_ring_zero_read_keep(m); +#endif + php_io_ring_req_release(ring, m); +#ifdef PHP_WIN32 + member_op->in_flight = zero_read_kept; +#endif +} + /* In a child the record can neither be cancelled nor complete: it is * settled here so that the release frees it */ static void php_io_ring_req_cancel_or_forget(php_io_ring *ring, php_io_ring_req *req) @@ -1633,7 +1692,7 @@ PHPAPI zend_result php_io_ring_cancel(php_io_ring *ring, php_io_op *op) php_io_ring_req *m = req->members[i]; m->group = NULL; php_io_ring_req_cancel_or_forget(ring, m); - php_io_ring_req_release(ring, m); + php_io_ring_member_release(ring, m); } req->n_members = 0; } else { @@ -1680,11 +1739,38 @@ PHPAPI bool php_io_ring_orphan(php_io_ring *ring, php_io_op *op) /* Completion processing */ +#ifdef PHP_WIN32 +/* A pipe POLL's NOP or zero-byte read as the IOR_POLL_* mask a poll would give */ +static int32_t php_io_ring_pipe_poll_res(const php_io_ring_req *req, int32_t res) +{ + if (res >= 0) { + return (int32_t) ((req->pipe_events & PHP_POLL_WRITE) ? IOR_POLL_OUT : IOR_POLL_IN); + } + + if (res == -EPIPE) { + return (int32_t) (IOR_POLL_IN | IOR_POLL_HUP); + } + + if (res == -ETIME || res == -ECANCELED) { + return res; + } + + /* A read set's write end fails with ERROR_ACCESS_DENIED: an error, as a socket's is reported */ + return (int32_t) IOR_POLL_ERR; +} +#endif + static void php_io_ring_result_from_cqe(php_io_ring_req *req, int32_t res) { php_io_op_result *r = &req->result; bool dns = req->type == PHP_IO_OP_GETADDRINFO || req->type == PHP_IO_OP_GETNAMEINFO; +#ifdef PHP_WIN32 + if (req->pipe_events) { + res = php_io_ring_pipe_poll_res(req, res); + } +#endif + r->index = req->index; r->error = 0; r->res = res; @@ -1788,6 +1874,40 @@ static void php_io_ring_req_lt_cqe(php_io_ring *ring, php_io_ring_req *req, int3 php_io_ring_req_complete(ring, req); } +#ifdef PHP_WIN32 +/* A pipe member is ready by the bytes it holds, answered as its zero-byte read would be: WSAPoll + * takes sockets only */ +static bool php_io_ring_pipe_peek(php_io_ring_req *m) +{ + int32_t res = 0; + if (!(m->pipe_events & PHP_POLL_WRITE)) { + DWORD avail = 0; + if (PeekNamedPipe(php_io_ring_file_fd(m->op), NULL, 0, NULL, &avail, NULL)) { + if (avail == 0) { + return false; + } + } else { + res = GetLastError() == ERROR_BROKEN_PIPE ? -EPIPE : -EIO; + } + } + + php_io_ring_result_from_cqe(m, res); + return true; +} +#endif + +/* A member the probe asks php_poll2() about */ +static zend_always_inline bool php_io_ring_probe_polls(const php_io_ring_req *m) +{ +#ifdef PHP_WIN32 + if (m->pipe_events) { + return false; + } +#endif + + return !m->main_done && m->type == PHP_IO_OP_POLL; +} + /* Poll members that are ready by now but whose cqe is not there yet (the * thread backend posts each from its poller in turn) are reported too */ static void php_io_ring_group_probe(php_io_ring_req *req, bool *probed) @@ -1798,7 +1918,13 @@ static void php_io_ring_group_probe(php_io_ring_req *req, bool *probed) for (uint32_t i = 0; i < req->n_members; i++) { php_io_ring_req *m = req->members[i]; - if (!m->main_done && m->type == PHP_IO_OP_POLL) { +#ifdef PHP_WIN32 + if (!m->main_done && m->pipe_events) { + probed[i] = php_io_ring_pipe_peek(m); + continue; + } +#endif + if (php_io_ring_probe_polls(m)) { n++; } } @@ -1811,7 +1937,7 @@ static void php_io_ring_group_probe(php_io_ring_req *req, bool *probed) n = 0; for (uint32_t i = 0; i < req->n_members; i++) { php_io_ring_req *m = req->members[i]; - if (!m->main_done && m->type == PHP_IO_OP_POLL) { + if (php_io_ring_probe_polls(m)) { fds[n].fd = m->op->fd; fds[n].events = (m->op->u.poll.events & PHP_POLL_READ ? POLLIN : 0) | (m->op->u.poll.events & PHP_POLL_WRITE ? POLLOUT : 0); @@ -1823,7 +1949,7 @@ static void php_io_ring_group_probe(php_io_ring_req *req, bool *probed) n = 0; for (uint32_t i = 0; i < req->n_members; i++) { php_io_ring_req *m = req->members[i]; - if (!m->main_done && m->type == PHP_IO_OP_POLL) { + if (php_io_ring_probe_polls(m)) { short revents = fds[n++].revents; if (revents) { uint32_t mask = (revents & POLLIN ? IOR_POLL_IN : 0) | (revents & POLLOUT ? IOR_POLL_OUT : 0) @@ -1865,7 +1991,7 @@ static void php_io_ring_group_fold(php_io_ring *ring, php_io_ring_req *req) php_io_ring_req_cancel(ring, m); } m->group = NULL; - php_io_ring_req_release(ring, m); + php_io_ring_member_release(ring, m); } req->n_members = 0; op->u.any.n_results = n_results; @@ -2042,6 +2168,12 @@ PHPAPI void php_io_ring_drain(php_io_ring *ring, const void *owner) pending = true; break; } +#ifdef PHP_WIN32 + if (r->zero_read_stream == owner) { + pending = true; + break; + } +#endif } if (!pending) { return; @@ -2368,6 +2500,16 @@ static uint32_t php_io_ring_queue_hook_flags(php_io_queue *base) return php_io_ring_hook_flags(((php_io_ring_queue *) base)->ring); } +#ifdef PHP_WIN32 +/* A failure (an op of the Ring still on the handle, a system before Windows 8.1) leaves it on the + * port, where ior drops the other process's packets */ +static void php_io_ring_queue_release(php_io_queue *base, php_socket_t fd) +{ + php_io_ring *const ring = ((php_io_ring_queue *) base)->ring; + ior_release_handle(ring->ctx, (ior_fd_t) _get_osfhandle((int) fd)); +} +#endif + static void php_io_ring_queue_destroy(php_io_queue *base) { php_io_ring_queue *q = (php_io_ring_queue *) base; @@ -2389,6 +2531,9 @@ static const php_io_queue_ops php_io_ring_queue_ops = { .count_pending = php_io_ring_queue_count_pending, .hook_flags = php_io_ring_queue_hook_flags, .destroy = php_io_ring_queue_destroy, +#ifdef PHP_WIN32 + .release = php_io_ring_queue_release, +#endif }; PHPAPI php_io_ring *php_io_queue_ring(php_io_queue *q) diff --git a/main/php_io_hooks.h b/main/php_io_hooks.h index 4b4c24d45222..f64bfe038b43 100644 --- a/main/php_io_hooks.h +++ b/main/php_io_hooks.h @@ -48,6 +48,15 @@ typedef enum { /* POLL: the caller checked readiness at arm time (a zero-timeout poll() found none), so a queue * need not check a registered pair again; never answered from recorded readiness */ #define PHP_IO_OP_F_CHECKED 0x04 +/* READ, WRITE, POLL on Windows: fd is the CRT descriptor of an overlapped pipe + * (php_io_overlapped_pipes), which no socket poll takes. A provider performs such an op, whatever + * its data flags, or answers Unsupported. A READ completes with the bytes there as soon as there + * are any, and with 0 or EPIPE at the writer's close; a WRITE to a closed reader fails with EPIPE. + * A POLL READ is ready at data, READ | HUP at the writer's close, ERROR on a write end; a POLL + * WRITE is ready at once, since a pipe has no write readiness. A POLL READ member of an Any may + * come back with in_flight set: its zero-byte read is still in the kernel, and the provider must + * drain its stream (the queue's drain) before run() returns. */ +#define PHP_IO_OP_F_PIPE 0x10 typedef enum { PHP_IO_DONE, @@ -255,6 +264,16 @@ PHPAPI extern bool (*php_io_signal_pending)(void); /* Whether a wait woken by a signal gives up rather than restarts */ PHPAPI bool php_io_interrupt_pending(void); +#ifdef PHP_WIN32 +/* Set at MINIT by the extension that installs the provider: proc_open() then makes overlapped + * named pipes, whose ops a provider may perform (PHP_IO_OP_F_PIPE), and php_select() judges every + * pipe by its bytes. False keeps anonymous pipes. */ +PHPAPI extern bool php_io_overlapped_pipes; +/* An overlapped pipe's buffer each way, and the most one write through a provider moves: a + * cancelled write leaves the reader an unknown part of it */ +#define PHP_IO_PIPE_BUFFER_SIZE (64 * 1024) +#endif + /* Without a provider it waits itself */ PHPAPI zend_result php_io_run(php_io_op *op, php_io_op_result *result); @@ -303,6 +322,20 @@ PHPAPI ssize_t php_io_read_at(php_stream *stream, int fd, void *buf, size_t len, php_deadline *dl); PHPAPI ssize_t php_io_write_at(php_stream *stream, int fd, const void *buf, size_t len, int64_t offset, php_deadline *dl); +#ifdef PHP_WIN32 +/* An overlapped pipe (PHP_IO_OP_F_PIPE). A read returns the bytes there, waiting for some unless dl + * is non-blocking (EAGAIN; any other dl waits without limit); 0 is the writer's close. A write + * through the provider moves at most PHP_IO_PIPE_BUFFER_SIZE, the thread's write the whole count; a + * closed reader is EPIPE. Without hooks, or when the provider answers Unsupported, both block the + * thread. no_provider: the descriptor was handed to another process. */ +PHPAPI ssize_t php_io_pipe_read(php_stream *stream, int fd, void *buf, size_t len, + const php_deadline *dl, bool no_provider); +PHPAPI ssize_t php_io_pipe_write(php_stream *stream, int fd, const void *buf, size_t len, + bool no_provider); +/* What a select reports for a pipe in its read set: bytes in it, or a peek failure (the writer's + * close, a write end) that the read reports */ +PHPAPI bool php_io_pipe_readable(int fd); +#endif PHPAPI php_socket_t php_io_accept(php_stream *stream, php_socket_t fd, struct sockaddr *addr, socklen_t *addrlen, php_deadline *dl); PHPAPI int php_io_connect(php_stream *stream, php_socket_t fd, const struct sockaddr *addr, @@ -344,6 +377,11 @@ PHPAPI void php_io_child_forget(pid_t pid); PHPAPI void php_io_stream_orphan(php_stream *stream, php_io_queue *queue); PHPAPI void php_io_stream_unfreeze(php_stream *stream); PHPAPI void php_io_stream_drain(php_stream *stream); +#ifdef PHP_WIN32 +/* php_io_stream_drain() for a stream that lives on: its descriptor goes to the thread or another + * process */ +PHPAPI void php_io_stream_settle(php_stream *stream); +#endif PHPAPI void php_io_handle_orphan(zend_object *handle, php_io_queue *queue); PHPAPI void php_io_handle_unfreeze(zend_object *handle); PHPAPI bool php_io_handle_busy(zend_object *handle); @@ -414,6 +452,9 @@ typedef struct _php_io_queue_ops { uint32_t (*count_pending)(php_io_queue *q); uint32_t (*hook_flags)(php_io_queue *q); void (*destroy)(php_io_queue *q); + /* A descriptor the queue's Read and Write ops use is about to go to another process: the queue's + * backend lets go of it unless an op of the queue is still on it. May be NULL. */ + void (*release)(php_io_queue *q, php_socket_t fd); } php_io_queue_ops; struct _php_io_queue { @@ -431,6 +472,8 @@ PHPAPI uint64_t php_io_queue_new_id(void); PHPAPI void php_io_queue_attach(php_io_queue *q, uint64_t id); PHPAPI void php_io_queue_detach(php_io_queue *q); PHPAPI php_io_queue *php_io_queue_find(uint64_t id); +/* release() on every queue of the thread */ +PHPAPI void php_io_queues_release(php_socket_t fd); /* The events among those asked that the queues of a registrant's pairs hold themselves, for a * readiness check made with a syscall */ PHPAPI uint32_t php_io_held_events(php_io_registration *regs, uint32_t events); diff --git a/main/php_streams.h b/main/php_streams.h index 9b7d122cb213..492dd48888be 100644 --- a/main/php_streams.h +++ b/main/php_streams.h @@ -527,6 +527,17 @@ END_EXTERN_C() #define php_stream_sync_supported(stream) (php_stream_set_option((stream), PHP_STREAM_OPTION_SYNC_API, PHP_STREAM_SYNC_SUPPORTED, NULL) == PHP_STREAM_OPTION_RETURN_OK ? 1 : 0) +/* Windows: OK for an overlapped proc_open() pipe (php_io_overlapped_pipes), else NOTIMPL. + * QUERY: stores in the bool ptrparam whether the descriptor was handed out. + * HAND_OUT: another process got the descriptor; the stream's reads and writes stop using a provider. + * RELEASE: just before the process is created, the queues let go of the descriptor, so the child's + * overlapped I/O does not post to their port. ERR (Error thrown) while another call has an op on + * the stream; an orphaned op is waited for first. */ +#define PHP_STREAM_OPTION_OVERLAPPED_PIPE 15 +#define PHP_STREAM_OVERLAPPED_PIPE_QUERY 0 +#define PHP_STREAM_OVERLAPPED_PIPE_HAND_OUT 1 +#define PHP_STREAM_OVERLAPPED_PIPE_RELEASE 2 + #define PHP_STREAM_OPTION_RETURN_OK 0 /* option set OK */ #define PHP_STREAM_OPTION_RETURN_ERR -1 /* problem setting option */ diff --git a/main/streams/php_stream_plain_wrapper.h b/main/streams/php_stream_plain_wrapper.h index a36897a3ede0..6d0edcd43268 100644 --- a/main/streams/php_stream_plain_wrapper.h +++ b/main/streams/php_stream_plain_wrapper.h @@ -33,6 +33,12 @@ PHPAPI php_stream *_php_stream_fopen_from_fd(int fd, const char *mode, const cha PHPAPI php_stream *_php_stream_fopen_from_pipe(FILE *file, const char *mode STREAMS_DC); #define php_stream_fopen_from_pipe(file, mode) _php_stream_fopen_from_pipe((file), (mode) STREAMS_CC) +#ifdef PHP_WIN32 +/* The parent's end of an overlapped named pipe proc_open() made (php_io_overlapped_pipes) */ +PHPAPI php_stream *_php_stream_fopen_from_overlapped_pipe(int fd, const char *mode STREAMS_DC); +#define php_stream_fopen_from_overlapped_pipe(fd, mode) _php_stream_fopen_from_overlapped_pipe((fd), (mode) STREAMS_CC) +#endif + #ifndef PHP_WIN32 /* popen(3) keeping the child's pid, so that closing the stream waits for it as an operation */ PHPAPI php_stream *_php_stream_popen(const char *command, const char *mode STREAMS_DC); diff --git a/main/streams/plain_wrapper.c b/main/streams/plain_wrapper.c index a855fdd10ab3..ce234a1993e3 100644 --- a/main/streams/plain_wrapper.c +++ b/main/streams/plain_wrapper.c @@ -144,7 +144,10 @@ typedef struct { unsigned is_seekable:1; /* don't try and seek, if not set */ unsigned is_overlapped:1; /* Windows: FILE_FLAG_OVERLAPPED, uses position */ unsigned nonblock_ours:1; /* O_NONBLOCK set for the IO hooks */ - unsigned _reserved:24; + unsigned is_overlapped_pipe:1; /* Windows: proc_open()'s overlapped named pipe (php_io_pipe_read) */ + unsigned overlapped_pipe_nonblock:1; /* Windows overlapped pipe: stream_set_blocking(false) */ + unsigned is_handed_out:1; /* Windows overlapped pipe: another process got it, the provider never sees it again */ + unsigned _reserved:21; #ifdef PHP_WIN32 zend_off_t position; /* of the next overlapped read or write */ int open_flags; /* the open()-style flags, for the synchronous reopen */ @@ -200,6 +203,20 @@ static int php_stdiop_make_sync(php_stdio_stream_data *data) data->is_overlapped = 0; return fd; } + +/* A synchronous call on an overlapped pipe may complete falsely while an overlapped one is pending + * on it (ReadFile()), so the descriptor goes out only when the stream has no op left */ +static zend_result php_stdiop_pipe_settle(php_stream *stream) +{ + if (php_io_stream_busy(stream)) { + zend_throw_error(NULL, "Concurrent access to a stream"); + return FAILURE; + } + + /* An orphaned op, already cancelled, is waited for */ + php_io_stream_settle(stream); + return SUCCESS; +} #endif /* The descriptor a cast hands out */ @@ -395,6 +412,19 @@ PHPAPI php_stream *_php_stream_fopen_from_fd(int fd, const char *mode, const cha return stream; } +#ifdef PHP_WIN32 +PHPAPI php_stream *_php_stream_fopen_from_overlapped_pipe(int fd, const char *mode STREAMS_DC) +{ + php_stream *const stream = php_stream_fopen_from_fd_rel(fd, mode, NULL, false); + + if (stream) { + ((php_stdio_stream_data*)stream->abstract)->is_overlapped_pipe = 1; + } + + return stream; +} +#endif + PHPAPI php_stream *_php_stream_fopen_from_file(FILE *file, const char *mode STREAMS_DC) { php_stream *stream = php_stream_fopen_from_file_int_rel(file, mode); @@ -561,7 +591,9 @@ static ssize_t php_stdiop_write(php_stream *stream, const char *buf, size_t coun if (data->fd >= 0) { #ifdef PHP_WIN32 - if (data->is_overlapped) { + if (data->is_overlapped_pipe) { + bytes_written = php_io_pipe_write(stream, data->fd, buf, PLAIN_WRAP_BUF_SIZE(count), data->is_handed_out); + } else if (data->is_overlapped) { php_deadline deadline; php_deadline_init_infinite(&deadline); bytes_written = php_io_write_at(stream, data->fd, buf, PLAIN_WRAP_BUF_SIZE(count), data->position, &deadline); @@ -625,7 +657,8 @@ static ssize_t php_stdiop_read(php_stream *stream, char *buf, size_t count) #ifdef PHP_WIN32 php_stdio_stream_data *self = (php_stdio_stream_data*)stream->abstract; - if ((self->is_pipe || self->is_process_pipe) && !self->is_pipe_blocking) { + /* An overlapped pipe waits for data in php_io_pipe_read() */ + if ((self->is_pipe || self->is_process_pipe) && !self->is_pipe_blocking && !self->is_overlapped_pipe) { HANDLE ph = (HANDLE)_get_osfhandle(data->fd); int retry = 0; DWORD avail_read = 0; @@ -651,7 +684,16 @@ static ssize_t php_stdiop_read(php_stream *stream, char *buf, size_t count) } #endif #ifdef PHP_WIN32 - if (data->is_overlapped) { + if (data->is_overlapped_pipe) { + php_deadline deadline; + if (data->overlapped_pipe_nonblock) { + php_deadline_init_nonblock(&deadline); + } else { + php_deadline_init_infinite(&deadline); + } + + ret = php_io_pipe_read(stream, data->fd, buf, PLAIN_WRAP_BUF_SIZE(count), &deadline, data->is_handed_out); + } else if (data->is_overlapped) { php_deadline deadline; php_deadline_init_infinite(&deadline); ret = php_io_read_at(stream, data->fd, buf, PLAIN_WRAP_BUF_SIZE(count), data->position, &deadline); @@ -907,6 +949,10 @@ static int php_stdiop_cast(php_stream *stream, int castas, void **ret) if (data->is_overlapped && php_stdiop_make_sync(data) < 0) { return FAILURE; } + + if (data->is_overlapped_pipe && php_stdiop_pipe_settle(stream) == FAILURE) { + return FAILURE; + } #endif if (data->file == NULL) { @@ -937,6 +983,12 @@ static int php_stdiop_cast(php_stream *stream, int castas, void **ret) return SUCCESS; case PHP_STREAM_AS_FD: +#ifdef PHP_WIN32 + if (ret && data->is_overlapped_pipe && php_stdiop_pipe_settle(stream) == FAILURE) { + return FAILURE; + } +#endif + fd = php_stdiop_cast_fd(data); if (SOCK_ERR == fd) { return FAILURE; @@ -1008,6 +1060,13 @@ static int php_stdiop_set_option(php_stream *stream, int option, int value, void case PHP_STREAM_OPTION_BLOCKING: if (fd == -1) return -1; +#ifdef PHP_WIN32 + if (data->is_overlapped_pipe) { + const int was_blocking = !data->overlapped_pipe_nonblock; + data->overlapped_pipe_nonblock = !value; + return was_blocking; + } +#endif #ifdef O_NONBLOCK if (value && data->nonblock_ours) { return 1; @@ -1312,6 +1371,26 @@ static int php_stdiop_set_option(php_stream *stream, int option, int value, void case PHP_STREAM_OPTION_PIPE_BLOCKING: data->is_pipe_blocking = value; return PHP_STREAM_OPTION_RETURN_OK; + + case PHP_STREAM_OPTION_OVERLAPPED_PIPE: + if (!data->is_overlapped_pipe) { + return PHP_STREAM_OPTION_RETURN_NOTIMPL; + } + + if (value == PHP_STREAM_OVERLAPPED_PIPE_HAND_OUT) { + data->is_handed_out = 1; + } else if (value == PHP_STREAM_OVERLAPPED_PIPE_RELEASE) { + /* An op still on the stream would tie the descriptor to the port again */ + if (php_stdiop_pipe_settle(stream) == FAILURE) { + return PHP_STREAM_OPTION_RETURN_ERR; + } + + php_io_queues_release(fd); + } else if (value == PHP_STREAM_OVERLAPPED_PIPE_QUERY && ptrparam) { + *(bool *) ptrparam = data->is_handed_out; + } + + return PHP_STREAM_OPTION_RETURN_OK; #endif case PHP_STREAM_OPTION_META_DATA_API: if (fd == -1) diff --git a/win32/ioutil.c b/win32/ioutil.c index ad49698d07b4..fc0aa8e7e708 100644 --- a/win32/ioutil.c +++ b/win32/ioutil.c @@ -1428,6 +1428,13 @@ static ssize_t php_win32_ioutil_overlapped_io(int fd, void *buf, size_t len, int if (ok) { return (ssize_t) done; } + + if (write_op && (err == ERROR_BROKEN_PIPE || err == ERROR_NO_DATA)) { + /* A pipe whose reader is gone: pending at the close, or started after it */ + _set_errno(EPIPE); + return -1; + } + if (err == ERROR_HANDLE_EOF || err == ERROR_BROKEN_PIPE) { return 0; } diff --git a/win32/ioutil.h b/win32/ioutil.h index 81a17a041b6b..56beeb85646e 100644 --- a/win32/ioutil.h +++ b/win32/ioutil.h @@ -339,6 +339,7 @@ zend_always_inline static int php_win32_ioutil_access(const char *path, mode_t m * with errno on failure. The completion never reaches a completion port * the handle is bound to. */ PW32IO ssize_t php_win32_ioutil_pread(int fd, void *buf, size_t len, int64_t offset); +/* A write to a pipe whose reader closed fails with EPIPE */ PW32IO ssize_t php_win32_ioutil_pwrite(int fd, const void *buf, size_t len, int64_t offset); zend_always_inline static int php_win32_ioutil_open(const char *path, int flags, ...) diff --git a/win32/select.c b/win32/select.c index 988f037d7a4c..ddafd5e62c06 100644 --- a/win32/select.c +++ b/win32/select.c @@ -14,6 +14,7 @@ #include "php.h" #include "php_network.h" +#include "php_io_hooks.h" #include "win32/time.h" /* Win32 select() will only work with sockets, so we roll our own implementation here. @@ -41,6 +42,13 @@ PHPAPI int php_select(php_socket_t max_fd, fd_set *rfds, fd_set *wfds, fd_set *e int sock_max_fd = -1; struct timeval tvslice; int retcode; + /* With php_io_overlapped_pipes every pipe is judged by its bytes: an overlapped pipe's handle + * is signaled whatever it holds, so it stays out of the wait; a pipe is always writable and, + * as a signaled handle was, always in the except set. Each set holds up to FD_SETSIZE. */ + int pipe_fds[3 * FD_SETSIZE]; + int n_pipes = 0, pipes_ready; + fd_set pipe_read, pipe_write, pipe_except; + DWORD slice_ms; /* As max_fd is unsigned, non socket might overflow. */ if (max_fd > (php_socket_t)INT_MAX) { @@ -83,7 +91,10 @@ PHPAPI int php_select(php_socket_t max_fd, fd_set *rfds, fd_set *wfds, fd_set *e } } else { handles[n_handles] = (HANDLE)(uintptr_t)_get_osfhandle(i); - if (handles[n_handles] != INVALID_HANDLE_VALUE) { + if (php_io_overlapped_pipes && handles[n_handles] != INVALID_HANDLE_VALUE + && GetFileType(handles[n_handles]) == FILE_TYPE_PIPE) { + pipe_fds[n_pipes++] = i; + } else if (handles[n_handles] != INVALID_HANDLE_VALUE) { if (SAFE_FD_ISSET(i, rfds) && GetFileType(handles[n_handles]) == FILE_TYPE_PIPE) { num_read_pipes++; } @@ -94,7 +105,7 @@ PHPAPI int php_select(php_socket_t max_fd, fd_set *rfds, fd_set *wfds, fd_set *e } } - if (n_handles == 0) { + if (n_handles == 0 && n_pipes == 0) { /* plain sockets only - let winsock handle the whole thing */ return select(-1, rfds, wfds, efds, tv); } @@ -110,6 +121,30 @@ PHPAPI int php_select(php_socket_t max_fd, fd_set *rfds, fd_set *wfds, fd_set *e do { retcode = 0; + pipes_ready = 0; + FD_ZERO(&pipe_read); + FD_ZERO(&pipe_write); + FD_ZERO(&pipe_except); + for (i = 0; i < n_pipes; i++) { + if (SAFE_FD_ISSET(pipe_fds[i], rfds) && php_io_pipe_readable(pipe_fds[i])) { + FD_SET((uint32_t)pipe_fds[i], &pipe_read); + pipes_ready++; + } + + if (SAFE_FD_ISSET(pipe_fds[i], wfds)) { + FD_SET((uint32_t)pipe_fds[i], &pipe_write); + pipes_ready++; + } + + if (SAFE_FD_ISSET(pipe_fds[i], efds)) { + FD_SET((uint32_t)pipe_fds[i], &pipe_except); + pipes_ready++; + } + } + + /* With pipes to peek again, the other waits are short; the timer tick rounds 1 ms up */ + slice_ms = pipes_ready > 0 ? 0 : (n_pipes > 0 ? 1 : 100); + if (sock_max_fd >= 0) { /* overwrite the zero'd sets here; the select call * will clear those that are not active */ @@ -118,7 +153,7 @@ PHPAPI int php_select(php_socket_t max_fd, fd_set *rfds, fd_set *wfds, fd_set *e aexcept = sock_except; tvslice.tv_sec = 0; - tvslice.tv_usec = 100000; + tvslice.tv_usec = slice_ms * 1000; retcode = select(-1, &aread, &awrite, &aexcept, &tvslice); } @@ -126,7 +161,7 @@ PHPAPI int php_select(php_socket_t max_fd, fd_set *rfds, fd_set *wfds, fd_set *e /* check handles */ DWORD wret; - wret = WaitForMultipleObjects(n_handles, handles, FALSE, retcode > 0 ? 0 : 100); + wret = WaitForMultipleObjects(n_handles, handles, FALSE, retcode > 0 ? 0 : slice_ms); if (wret == WAIT_TIMEOUT) { /* set retcode to 0; this is the default. @@ -166,6 +201,27 @@ PHPAPI int php_select(php_socket_t max_fd, fd_set *rfds, fd_set *wfds, fd_set *e } } } + + /* After a failed select(), aread and awrite still hold the socket sets, so the failure stays + * the answer */ + if (pipes_ready > 0 && retcode >= 0) { + for (i = 0; i < n_pipes; i++) { + if (FD_ISSET(pipe_fds[i], &pipe_read)) { + FD_SET((uint32_t)pipe_fds[i], &aread); + } + + if (FD_ISSET(pipe_fds[i], &pipe_write)) { + FD_SET((uint32_t)pipe_fds[i], &awrite); + } + + if (FD_ISSET(pipe_fds[i], &pipe_except)) { + FD_SET((uint32_t)pipe_fds[i], &aexcept); + } + } + + retcode += pipes_ready; + } + if (retcode == 0 && num_read_pipes == n_handles && sock_max_fd < 0) { usleep(100); }