[PATCH 0/2] MR135: Cancel asyncs queued by thread when the thread is terminated
Inspired by the discussion for MR 127 (https://www.winehq.org/pipermail/wine-devel/2022-May/217832.html) and is aimed to replace the second patch (ws2_32: Use allocated IO status block in select().) in that MR. When the thread calls server's terminate_thread for itself queuing the system APC though SIGUSR1 might race with the thread exit, there seems no easy and reliable way to be sure that signal is processed before the thread's stack and TEB are freed. Thus I am making the system APCs to be queued to the thread to be processed by server_select() after (terminate_thread) server call. When terminate_thread is called for the other thread the system APC will always be delivered through some other thread. Thread stack and TEB are not freed when the thread is killed externally, so completing the async from another thread should probably be fine. -- https://gitlab.winehq.org/wine/wine/-/merge_requests/135
From: Paul Gofman <pgofman(a)codeweavers.com> --- dlls/ws2_32/tests/afd.c | 159 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 159 insertions(+) diff --git a/dlls/ws2_32/tests/afd.c b/dlls/ws2_32/tests/afd.c index 5b64ccd7715..b8579a7c9f7 100644 --- a/dlls/ws2_32/tests/afd.c +++ b/dlls/ws2_32/tests/afd.c @@ -2355,6 +2355,164 @@ static void test_getsockname(void) CloseHandle(event); } +struct ioctl_params +{ + HANDLE handle, event; + PIO_APC_ROUTINE apc; + void *apc_context; + IO_STATUS_BLOCK *io; + ULONG code; + void *in_buffer; + ULONG in_size; + void *out_buffer; + ULONG out_size; + NTSTATUS ret; + HANDLE complete_event; + BOOL kill_thread; +}; + +static DWORD WINAPI async_ioctl_thread(void *params) +{ + struct ioctl_params *io = params; + + io->ret = NtDeviceIoControlFile(io->handle, io->event, io->apc, io->apc_context, io->io, + io->code, io->in_buffer, io->in_size, io->out_buffer, io->out_size); + SetEvent(io->complete_event); + if (io->kill_thread) + Sleep(3000); + return io->ret; +} + +static NTSTATUS WINAPI thread_NtDeviceIoControlFile(BOOL kill_thread, HANDLE handle, HANDLE event, + PIO_APC_ROUTINE apc, void *apc_context, IO_STATUS_BLOCK *io, ULONG code, void *in_buffer, ULONG in_size, + void *out_buffer, ULONG out_size) +{ + struct ioctl_params p; + HANDLE thread; + DWORD ret; + + p.handle = handle; + p.event = event; + p.apc = apc; + p.apc_context = apc_context; + p.io = io; + p.code = code; + p.in_buffer = in_buffer; + p.in_size = in_size; + p.out_buffer = out_buffer; + p.out_size = out_size; + p.complete_event = CreateEventW(NULL, FALSE, FALSE, NULL); + p.kill_thread = kill_thread; + + thread = CreateThread(NULL, 0, async_ioctl_thread, &p, 0, NULL); + ok(!!thread, "got NULL.\n"); + ret = WaitForSingleObject(p.complete_event, INFINITE); + ok(ret == WAIT_OBJECT_0, "got ret %#lx.\n", ret); + if (kill_thread) + TerminateThread(thread, -1); + CloseHandle(p.complete_event); + ret = WaitForSingleObject(thread, INFINITE); + ok(ret == WAIT_OBJECT_0, "got ret %#lx.\n", ret); + CloseHandle(thread); + return p.ret; +} + +static void test_async_thread_termination(void) +{ + const struct sockaddr_in bind_addr = {.sin_family = AF_INET, .sin_addr.s_addr = htonl(INADDR_LOOPBACK)}; + char in_buffer[offsetof(struct afd_poll_params, sockets[3])]; + char out_buffer[offsetof(struct afd_poll_params, sockets[3])]; + struct afd_poll_params *in_params = (struct afd_poll_params *)in_buffer; + struct afd_poll_params *out_params = (struct afd_poll_params *)out_buffer; + LARGE_INTEGER zero = {{0}}; + ULONG_PTR key, value; + IO_STATUS_BLOCK io; + HANDLE event, port; + ULONG params_size; + SOCKET listener; + int ret; + + event = CreateEventW(NULL, FALSE, FALSE, NULL); + + listener = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP); + ret = bind(listener, (const struct sockaddr *)&bind_addr, sizeof(bind_addr)); + ok(!ret, "got error %u\n", WSAGetLastError()); + ret = listen(listener, 1); + ok(!ret, "got error %u\n", WSAGetLastError()); + + in_params->count = 1; + in_params->exclusive = FALSE; + in_params->sockets[0].socket = listener; + in_params->sockets[0].flags = ~0; + in_params->sockets[0].status = 0xdeadbeef; + params_size = offsetof(struct afd_poll_params, sockets[1]); + in_params->timeout = -10 * 1000 * 1000 * 5; + + memset(&io, 0xcc, sizeof(io)); + ret = thread_NtDeviceIoControlFile(TRUE, (HANDLE)listener, event, NULL, NULL, &io, + IOCTL_AFD_POLL, in_params, params_size, out_params, params_size); + ok(ret == STATUS_PENDING, "got %#x\n", ret); + ret = WaitForSingleObject(event, 1000); + todo_wine ok(!ret, "got %#x\n", ret); + todo_wine ok(io.Status == STATUS_CANCELLED, "got %#lx\n", io.Status); + + memset(&io, 0xcc, sizeof(io)); + ret = thread_NtDeviceIoControlFile(FALSE, (HANDLE)listener, event, NULL, NULL, &io, + IOCTL_AFD_POLL, in_params, params_size, out_params, params_size); + ok(ret == STATUS_PENDING, "got %#x\n", ret); + ret = WaitForSingleObject(event, 1000); + todo_wine ok(!ret, "got %#x\n", ret); + todo_wine ok(io.Status == STATUS_CANCELLED, "got %#lx\n", io.Status); + + port = CreateIoCompletionPort((HANDLE)listener, NULL, 0, 0); + + memset(&io, 0xcc, sizeof(io)); + ret = thread_NtDeviceIoControlFile(TRUE, (HANDLE)listener, event, NULL, (void *)0xdeadbeef, &io, + IOCTL_AFD_POLL, in_params, params_size, out_params, params_size); + ok(ret == STATUS_PENDING, "got %#x\n", ret); + + ret = WaitForSingleObject(event, 1000); + todo_wine ok(!ret, "got %#x\n", ret); + todo_wine ok(io.Status == STATUS_CANCELLED, "got %#lx\n", io.Status); + + memset(&io, 0xcc, sizeof(io)); + key = 0xcc; + value = 0; + ret = NtRemoveIoCompletion(port, &key, &value, &io, &zero); + todo_wine + { + ok(!ret, "got %#x\n", ret); + ok(!key, "got key %#Ix\n", key); + ok(value == 0xdeadbeef, "got value %#Ix\n", value); + ok(io.Status == STATUS_CANCELLED, "got %#lx\n", io.Status); + } + + memset(&io, 0xcc, sizeof(io)); + ret = thread_NtDeviceIoControlFile(FALSE, (HANDLE)listener, event, NULL, (void *)0xdeadbeef, &io, + IOCTL_AFD_POLL, in_params, params_size, out_params, params_size); + ok(ret == STATUS_PENDING, "got %#x\n", ret); + + ret = WaitForSingleObject(event, 1000); + todo_wine ok(!ret, "got %#x\n", ret); + todo_wine ok(io.Status == STATUS_CANCELLED, "got %#lx\n", io.Status); + + memset(&io, 0xcc, sizeof(io)); + key = 0xcc; + value = 0; + ret = NtRemoveIoCompletion(port, &key, &value, &io, &zero); + todo_wine + { + ok(!ret, "got %#x\n", ret); + ok(!key, "got key %#Ix\n", key); + ok(value == 0xdeadbeef, "got value %#Ix\n", value); + ok(io.Status == STATUS_CANCELLED, "got %#lx\n", io.Status); + } + + CloseHandle(port); + CloseHandle(event); + closesocket(listener); +} + START_TEST(afd) { WSADATA data; @@ -2372,6 +2530,7 @@ START_TEST(afd) test_get_events_reset(); test_bind(); test_getsockname(); + test_async_thread_termination(); WSACleanup(); } -- GitLab https://gitlab.winehq.org/wine/wine/-/merge_requests/135
From: Paul Gofman <pgofman(a)codeweavers.com> --- dlls/ntdll/unix/thread.c | 6 +++++- dlls/ws2_32/tests/afd.c | 38 ++++++++++++++++---------------------- server/async.c | 19 ++++++++++++++----- server/file.h | 1 + server/thread.c | 14 ++++++++------ server/thread.h | 3 ++- server/timer.c | 2 +- 7 files changed, 47 insertions(+), 36 deletions(-) diff --git a/dlls/ntdll/unix/thread.c b/dlls/ntdll/unix/thread.c index 15bb3be34b2..c202897f172 100644 --- a/dlls/ntdll/unix/thread.c +++ b/dlls/ntdll/unix/thread.c @@ -1623,7 +1623,11 @@ NTSTATUS WINAPI NtTerminateThread( HANDLE handle, LONG exit_code ) } SERVER_END_REQ; } - if (self) exit_thread( exit_code ); + if (self) + { + server_select( NULL, 0, SELECT_INTERRUPTIBLE, 0, NULL, NULL ); + exit_thread( exit_code ); + } return ret; } diff --git a/dlls/ws2_32/tests/afd.c b/dlls/ws2_32/tests/afd.c index b8579a7c9f7..de042358297 100644 --- a/dlls/ws2_32/tests/afd.c +++ b/dlls/ws2_32/tests/afd.c @@ -2453,16 +2453,16 @@ static void test_async_thread_termination(void) IOCTL_AFD_POLL, in_params, params_size, out_params, params_size); ok(ret == STATUS_PENDING, "got %#x\n", ret); ret = WaitForSingleObject(event, 1000); - todo_wine ok(!ret, "got %#x\n", ret); - todo_wine ok(io.Status == STATUS_CANCELLED, "got %#lx\n", io.Status); + ok(!ret, "got %#x\n", ret); + ok(io.Status == STATUS_CANCELLED, "got %#lx\n", io.Status); memset(&io, 0xcc, sizeof(io)); ret = thread_NtDeviceIoControlFile(FALSE, (HANDLE)listener, event, NULL, NULL, &io, IOCTL_AFD_POLL, in_params, params_size, out_params, params_size); ok(ret == STATUS_PENDING, "got %#x\n", ret); ret = WaitForSingleObject(event, 1000); - todo_wine ok(!ret, "got %#x\n", ret); - todo_wine ok(io.Status == STATUS_CANCELLED, "got %#lx\n", io.Status); + ok(!ret, "got %#x\n", ret); + ok(io.Status == STATUS_CANCELLED, "got %#lx\n", io.Status); port = CreateIoCompletionPort((HANDLE)listener, NULL, 0, 0); @@ -2472,20 +2472,17 @@ static void test_async_thread_termination(void) ok(ret == STATUS_PENDING, "got %#x\n", ret); ret = WaitForSingleObject(event, 1000); - todo_wine ok(!ret, "got %#x\n", ret); - todo_wine ok(io.Status == STATUS_CANCELLED, "got %#lx\n", io.Status); + ok(!ret, "got %#x\n", ret); + ok(io.Status == STATUS_CANCELLED, "got %#lx\n", io.Status); memset(&io, 0xcc, sizeof(io)); key = 0xcc; value = 0; ret = NtRemoveIoCompletion(port, &key, &value, &io, &zero); - todo_wine - { - ok(!ret, "got %#x\n", ret); - ok(!key, "got key %#Ix\n", key); - ok(value == 0xdeadbeef, "got value %#Ix\n", value); - ok(io.Status == STATUS_CANCELLED, "got %#lx\n", io.Status); - } + ok(!ret, "got %#x\n", ret); + ok(!key, "got key %#Ix\n", key); + ok(value == 0xdeadbeef, "got value %#Ix\n", value); + ok(io.Status == STATUS_CANCELLED, "got %#lx\n", io.Status); memset(&io, 0xcc, sizeof(io)); ret = thread_NtDeviceIoControlFile(FALSE, (HANDLE)listener, event, NULL, (void *)0xdeadbeef, &io, @@ -2493,20 +2490,17 @@ static void test_async_thread_termination(void) ok(ret == STATUS_PENDING, "got %#x\n", ret); ret = WaitForSingleObject(event, 1000); - todo_wine ok(!ret, "got %#x\n", ret); - todo_wine ok(io.Status == STATUS_CANCELLED, "got %#lx\n", io.Status); + ok(!ret, "got %#x\n", ret); + ok(io.Status == STATUS_CANCELLED, "got %#lx\n", io.Status); memset(&io, 0xcc, sizeof(io)); key = 0xcc; value = 0; ret = NtRemoveIoCompletion(port, &key, &value, &io, &zero); - todo_wine - { - ok(!ret, "got %#x\n", ret); - ok(!key, "got key %#Ix\n", key); - ok(value == 0xdeadbeef, "got value %#Ix\n", value); - ok(io.Status == STATUS_CANCELLED, "got %#lx\n", io.Status); - } + ok(!ret, "got %#x\n", ret); + ok(!key, "got key %#Ix\n", key); + ok(value == 0xdeadbeef, "got value %#Ix\n", value); + ok(io.Status == STATUS_CANCELLED, "got %#lx\n", io.Status); CloseHandle(port); CloseHandle(event); diff --git a/server/async.c b/server/async.c index a4fbeab555e..46a7b06a0dd 100644 --- a/server/async.c +++ b/server/async.c @@ -62,6 +62,7 @@ struct async unsigned int comp_flags; /* completion flags */ async_completion_callback completion_callback; /* callback to be called on completion */ void *completion_callback_private; /* argument to completion_callback */ + int thread_terminating; /* async is being canceled due to thread termination */ }; static void async_dump( struct object *obj, int verbose ); @@ -202,7 +203,7 @@ void async_terminate( struct async *async, unsigned int status ) else data.async_io.status = status; - thread_queue_apc( async->thread->process, async->thread, &async->obj, &data ); + thread_queue_apc( async->thread->process, async->thread, &async->obj, &data, async->thread_terminating ); } async_reselect( async ); @@ -281,6 +282,7 @@ struct async *create_async( struct fd *fd, struct thread *thread, const async_da async->comp_flags = 0; async->completion_callback = NULL; async->completion_callback_private = NULL; + async->thread_terminating = 0; if (iosb) async->iosb = (struct iosb *)grab_object( iosb ); else async->iosb = NULL; @@ -520,7 +522,7 @@ void async_set_result( struct object *obj, unsigned int status, apc_param_t tota data.user.args[0] = async->data.apc_context; data.user.args[1] = async->data.iosb; data.user.args[2] = 0; - thread_queue_apc( NULL, async->thread, NULL, &data ); + thread_queue_apc( NULL, async->thread, NULL, &data, 0 ); } else if (async->data.apc_context && (async->pending || !(async->comp_flags & FILE_SKIP_COMPLETION_PORT_ON_SUCCESS))) @@ -562,7 +564,8 @@ int async_waiting( struct async_queue *queue ) return !async->terminated; } -static int cancel_async( struct process *process, struct object *obj, struct thread *thread, client_ptr_t iosb ) +static int cancel_async( struct process *process, struct object *obj, struct thread *thread, client_ptr_t iosb, + int thread_terminating ) { struct async *async; int woken = 0; @@ -580,6 +583,7 @@ restart: (!iosb || async->data.iosb == iosb)) { async->canceled = 1; + async->thread_terminating = thread_terminating; fd_cancel_async( async->fd, async ); woken++; goto restart; @@ -590,7 +594,12 @@ restart: void cancel_process_asyncs( struct process *process ) { - cancel_async( process, NULL, NULL, 0 ); + cancel_async( process, NULL, NULL, 0, 0 ); +} + +void cancel_terminating_thread_asyncs( struct thread *thread ) +{ + cancel_async( thread->process, NULL, thread, 0, 1 ); } /* wake up async operations on the queue */ @@ -723,7 +732,7 @@ DECL_HANDLER(cancel_async) if (obj) { - int count = cancel_async( current->process, obj, thread, req->iosb ); + int count = cancel_async( current->process, obj, thread, req->iosb, 0 ); if (!count && req->iosb) set_error( STATUS_NOT_FOUND ); release_object( obj ); } diff --git a/server/file.h b/server/file.h index 9f9d4cd4e1a..0ffe0e2c8dc 100644 --- a/server/file.h +++ b/server/file.h @@ -245,6 +245,7 @@ extern struct iosb *async_get_iosb( struct async *async ); extern struct thread *async_get_thread( struct async *async ); extern struct async *find_pending_async( struct async_queue *queue ); extern void cancel_process_asyncs( struct process *process ); +extern void cancel_terminating_thread_asyncs( struct thread *thread ); static inline void init_async_queue( struct async_queue *queue ) { diff --git a/server/thread.c b/server/thread.c index 467ccd1f0db..e7c0f294c55 100644 --- a/server/thread.c +++ b/server/thread.c @@ -1094,7 +1094,7 @@ static inline int is_in_apc_wait( struct thread *thread ) } /* queue an existing APC to a given thread */ -static int queue_apc( struct process *process, struct thread *thread, struct thread_apc *apc ) +static int queue_apc( struct process *process, struct thread *thread, struct thread_apc *apc, int queue_only ) { struct list *queue; @@ -1135,7 +1135,7 @@ static int queue_apc( struct process *process, struct thread *thread, struct thr if (thread->state == TERMINATED) return 0; if (!(queue = get_apc_queue( thread, apc->call.type ))) return 1; /* send signal for system APCs if needed */ - if (queue == &thread->system_apc && list_empty( queue ) && !is_in_apc_wait( thread )) + if (!queue_only && queue == &thread->system_apc && list_empty( queue ) && !is_in_apc_wait( thread )) { if (!send_thread_signal( thread, SIGUSR1 )) return 0; } @@ -1152,14 +1152,15 @@ static int queue_apc( struct process *process, struct thread *thread, struct thr } /* queue an async procedure call */ -int thread_queue_apc( struct process *process, struct thread *thread, struct object *owner, const apc_call_t *call_data ) +int thread_queue_apc( struct process *process, struct thread *thread, struct object *owner, + const apc_call_t *call_data, int queue_only ) { struct thread_apc *apc; int ret = 0; if ((apc = create_apc( owner, call_data ))) { - ret = queue_apc( process, thread, apc ); + ret = queue_apc( process, thread, apc, queue_only ); release_object( apc ); } return ret; @@ -1462,6 +1463,7 @@ DECL_HANDLER(terminate_thread) thread->exit_code = req->exit_code; if (thread != current) kill_thread( thread, 1 ); else reply->self = 1; + cancel_terminating_thread_asyncs( thread ); release_object( thread ); } } @@ -1761,7 +1763,7 @@ DECL_HANDLER(queue_apc) if (thread) { - if (!queue_apc( NULL, thread, apc )) set_error( STATUS_UNSUCCESSFUL ); + if (!queue_apc( NULL, thread, apc, 0 )) set_error( STATUS_UNSUCCESSFUL ); release_object( thread ); } else if (process) @@ -1772,7 +1774,7 @@ DECL_HANDLER(queue_apc) obj_handle_t handle = alloc_handle( current->process, apc, SYNCHRONIZE, 0 ); if (handle) { - if (queue_apc( process, NULL, apc )) + if (queue_apc( process, NULL, apc, 0 )) { apc->caller = (struct thread *)grab_object( current ); reply->handle = handle; diff --git a/server/thread.h b/server/thread.h index 8dcf966a90a..9c84c5133b9 100644 --- a/server/thread.h +++ b/server/thread.h @@ -114,7 +114,8 @@ extern int add_queue( struct object *obj, struct wait_queue_entry *entry ); extern void remove_queue( struct object *obj, struct wait_queue_entry *entry ); extern void kill_thread( struct thread *thread, int violent_death ); extern void wake_up( struct object *obj, int max ); -extern int thread_queue_apc( struct process *process, struct thread *thread, struct object *owner, const apc_call_t *call_data ); +extern int thread_queue_apc( struct process *process, struct thread *thread, struct object *owner, + const apc_call_t *call_data, int queue_only ); extern void thread_cancel_apc( struct thread *thread, struct object *owner, enum apc_type type ); extern int thread_add_inflight_fd( struct thread *thread, int client, int server ); extern int thread_get_inflight_fd( struct thread *thread, int client ); diff --git a/server/timer.c b/server/timer.c index 96dc9d00ca1..98a340c789f 100644 --- a/server/timer.c +++ b/server/timer.c @@ -133,7 +133,7 @@ static void timer_callback( void *private ) data.user.args[1] = (unsigned int)timer->when; data.user.args[2] = timer->when >> 32; - if (!thread_queue_apc( NULL, timer->thread, &timer->obj, &data )) + if (!thread_queue_apc( NULL, timer->thread, &timer->obj, &data, 0 )) { release_object( timer->thread ); timer->thread = NULL; -- GitLab https://gitlab.winehq.org/wine/wine/-/merge_requests/135
participants (2)
-
Paul Gofman -
Paul Gofman (＠gofman)