[PATCH 0/6] MR9086: server: Continuously poll on queue fd for driver events.
Allowing to use a simple event-like sync for the message queues. -- https://gitlab.winehq.org/wine/wine/-/merge_requests/9086
From: Rémi Bernon <rbernon(a)codeweavers.com> Instead of relying on ProcessEvents return value. --- dlls/win32u/message.c | 18 +++++++++++++++++- 1 file changed, 17 insertions(+), 1 deletion(-) diff --git a/dlls/win32u/message.c b/dlls/win32u/message.c index dc316922581..87efad6202c 100644 --- a/dlls/win32u/message.c +++ b/dlls/win32u/message.c @@ -3164,9 +3164,25 @@ static HANDLE get_server_queue_handle(void) return ret; } +static BOOL is_queue_signaled(void) +{ + struct object_lock lock = OBJECT_LOCK_INIT; + const queue_shm_t *queue_shm; + BOOL signaled = FALSE; + UINT status; + + while ((status = get_shared_queue( &lock, &queue_shm )) == STATUS_PENDING) + signaled = (queue_shm->wake_bits & queue_shm->wake_mask) || + (queue_shm->changed_bits & queue_shm->changed_mask); + if (status) return FALSE; + + return signaled; +} + BOOL process_driver_events( UINT mask ) { - return user_driver->pProcessEvents( mask ); + user_driver->pProcessEvents( mask ); + return is_queue_signaled(); } void check_for_events( UINT flags ) -- GitLab https://gitlab.winehq.org/wine/wine/-/merge_requests/9086
From: Rémi Bernon <rbernon(a)codeweavers.com> --- dlls/wineandroid.drv/window.c | 11 +++++++--- dlls/winemac.drv/event.c | 34 ++++++++++++++--------------- dlls/winex11.drv/event.c | 40 ++++++++++++++++------------------- 3 files changed, 43 insertions(+), 42 deletions(-) diff --git a/dlls/wineandroid.drv/window.c b/dlls/wineandroid.drv/window.c index 9c0dc8d6115..81be65a065c 100644 --- a/dlls/wineandroid.drv/window.c +++ b/dlls/wineandroid.drv/window.c @@ -391,13 +391,19 @@ static void pull_events(void) } +static int check_fd_events( int fd, int events ) +{ + struct pollfd pfd = {.fd = fd, .events = events}; + if (poll( &pfd, 1, 0 ) <= 0) return 0; + return pfd.revents; +} + /*********************************************************************** * process_events */ static int process_events( DWORD mask ) { struct java_event *event, *next, *previous; - unsigned int count = 0; assert( GetCurrentThreadId() == desktop_tid ); @@ -505,12 +511,11 @@ static int process_events( DWORD mask ) FIXME( "got event %u\n", event->data.type ); } free( event ); - count++; /* next may have been removed by a recursive call, so reset it to the beginning of the list */ next = LIST_ENTRY( event_queue.next, struct java_event, entry ); } current_event = previous; - return count; + return !check_fd_events( event_pipe[0], POLLIN ); } diff --git a/dlls/winemac.drv/event.c b/dlls/winemac.drv/event.c index f86716c7942..b97b876cb70 100644 --- a/dlls/winemac.drv/event.c +++ b/dlls/winemac.drv/event.c @@ -26,6 +26,8 @@ #include "config.h" +#include <poll.h> + #include "ntstatus.h" #define WIN32_NO_STATUS #include "macdrv.h" @@ -526,25 +528,13 @@ void macdrv_handle_event(const macdrv_event *event) } -/*********************************************************************** - * process_events - */ -static int process_events(macdrv_event_queue queue, macdrv_event_mask mask) +static int check_fd_events( int fd, int events ) { - macdrv_event *event; - int count = 0; - - while (macdrv_copy_event_from_queue(queue, mask, &event)) - { - count++; - macdrv_handle_event(event); - macdrv_release_event(event); - } - if (count) TRACE("processed %d events\n", count); - return count; + struct pollfd pfd = {.fd = fd, .events = events}; + if (poll( &pfd, 1, 0 ) <= 0) return 0; + return pfd.revents; } - /*********************************************************************** * ProcessEvents (MACDRV.@) */ @@ -552,6 +542,8 @@ BOOL macdrv_ProcessEvents(DWORD mask) { struct macdrv_thread_data *data = macdrv_thread_data(); macdrv_event_mask event_mask = get_event_mask(mask); + macdrv_event *event; + int count = 0; TRACE("mask %x\n", mask); @@ -563,5 +555,13 @@ BOOL macdrv_ProcessEvents(DWORD mask) data->current_event->type != WINDOW_DRAG_BEGIN) event_mask = 0; /* don't process nested events */ - return process_events(data->queue, event_mask); + while (macdrv_copy_event_from_queue(data->queue, event_mask, &event)) + { + count++; + macdrv_handle_event(event); + macdrv_release_event(event); + } + + if (count) TRACE("processed %d events\n", count); + return !check_fd_events(macdrv_get_event_queue_fd(data->queue), POLLIN); } diff --git a/dlls/winex11.drv/event.c b/dlls/winex11.drv/event.c index d2cf9e55977..ca261b7bf3b 100644 --- a/dlls/winex11.drv/event.c +++ b/dlls/winex11.drv/event.c @@ -464,19 +464,28 @@ static inline BOOL call_event_handler( Display *display, XEvent *event ) return ret; } +static int check_fd_events( int fd, int events ) +{ + struct pollfd pfd = {.fd = fd, .events = events}; + if (poll( &pfd, 1, 0 ) <= 0) return 0; + return pfd.revents; +} /*********************************************************************** - * process_events + * ProcessEvents (X11DRV.@) */ -static BOOL process_events( Display *display, Bool (*filter)(Display*, XEvent*,XPointer), ULONG_PTR arg ) +BOOL X11DRV_ProcessEvents( DWORD mask ) { + struct x11drv_thread_data *data = x11drv_thread_data(); XEvent event, prev_event; int count = 0; - BOOL queued = FALSE; enum event_merge_action action = MERGE_DISCARD; + if (!data) return FALSE; + if (data->current_event) mask = 0; /* don't process nested events */ + prev_event.type = 0; - while (XCheckIfEvent( display, &event, filter, (char *)arg )) + while (XCheckIfEvent( data->display, &event, filter_event, (XPointer)(UINT_PTR)mask )) { count++; if (XFilterEvent( &event, None )) @@ -517,41 +526,28 @@ static BOOL process_events( Display *display, Bool (*filter)(Display*, XEvent*,X switch( action ) { case MERGE_HANDLE: /* handle prev, keep new */ - queued |= call_event_handler( display, &prev_event ); + call_event_handler( data->display, &prev_event ); /* fall through */ case MERGE_DISCARD: /* discard prev, keep new */ free_event_data( &prev_event ); prev_event = event; break; case MERGE_KEEP: /* handle new, keep prev for future merging */ - queued |= call_event_handler( display, &event ); + call_event_handler( data->display, &event ); /* fall through */ case MERGE_IGNORE: /* ignore new, keep prev for future merging */ free_event_data( &event ); break; } } - if (prev_event.type) queued |= call_event_handler( display, &prev_event ); + if (prev_event.type) call_event_handler( data->display, &prev_event ); free_event_data( &prev_event ); XFlush( gdi_display ); - if (count) TRACE( "processed %d events, returning %d\n", count, queued ); - return queued; + if (count) TRACE( "processed %d events\n", count ); + return !check_fd_events( ConnectionNumber( data->display ), POLLIN ); } -/*********************************************************************** - * ProcessEvents (X11DRV.@) - */ -BOOL X11DRV_ProcessEvents( DWORD mask ) -{ - struct x11drv_thread_data *data = x11drv_thread_data(); - - if (!data) return FALSE; - if (data->current_event) mask = 0; /* don't process nested events */ - - return process_events( data->display, filter_event, mask ); -} - /*********************************************************************** * EVENT_x11_time_to_win32_time * -- GitLab https://gitlab.winehq.org/wine/wine/-/merge_requests/9086
From: Rémi Bernon <rbernon(a)codeweavers.com> --- dlls/win32u/message.c | 11 ++++++++++- server/protocol.def | 1 + server/queue.c | 8 ++++++++ 3 files changed, 19 insertions(+), 1 deletion(-) diff --git a/dlls/win32u/message.c b/dlls/win32u/message.c index 87efad6202c..02f937715bf 100644 --- a/dlls/win32u/message.c +++ b/dlls/win32u/message.c @@ -3181,7 +3181,16 @@ static BOOL is_queue_signaled(void) BOOL process_driver_events( UINT mask ) { - user_driver->pProcessEvents( mask ); + if (user_driver->pProcessEvents( mask )) + { + SERVER_START_REQ( set_queue_mask ) + { + req->poll_events = 1; + wine_server_call( req ); + } + SERVER_END_REQ; + } + return is_queue_signaled(); } diff --git a/server/protocol.def b/server/protocol.def index e0bd2ee5a07..4fc12225fdb 100644 --- a/server/protocol.def +++ b/server/protocol.def @@ -2302,6 +2302,7 @@ struct process_info @REQ(set_queue_mask) unsigned int wake_mask; /* wakeup bits mask */ unsigned int changed_mask; /* changed bits mask */ + int poll_events; /* whether to poll queue fd */ @REPLY unsigned int wake_bits; /* current wake bits */ unsigned int changed_bits; /* current changed bits */ diff --git a/server/queue.c b/server/queue.c index ffb1d441e60..2d36e1568c0 100644 --- a/server/queue.c +++ b/server/queue.c @@ -3158,6 +3158,14 @@ DECL_HANDLER(set_queue_mask) if (!queue) return; queue_shm = queue->shared; + if (req->poll_events) + { + if (!queue->fd) return; + if (check_fd_events( queue->fd, POLLIN )) set_queue_bits( queue, QS_DRIVER ); + else clear_queue_bits( queue, QS_DRIVER ); + return; + } + SHARED_WRITE_BEGIN( queue_shm, queue_shm_t ) { shared->access_time = monotonic_time; -- GitLab https://gitlab.winehq.org/wine/wine/-/merge_requests/9086
From: Rémi Bernon <rbernon(a)codeweavers.com> --- server/queue.c | 33 ++++++--------------------------- 1 file changed, 6 insertions(+), 27 deletions(-) diff --git a/server/queue.c b/server/queue.c index 2d36e1568c0..7ae58bc2b2e 100644 --- a/server/queue.c +++ b/server/queue.c @@ -152,7 +152,6 @@ struct hotkey static void msg_queue_dump( struct object *obj, int verbose ); static int msg_queue_add_queue( struct object *obj, struct wait_queue_entry *entry ); -static void msg_queue_remove_queue( struct object *obj, struct wait_queue_entry *entry ); static int msg_queue_signaled( struct object *obj, struct wait_queue_entry *entry ); static void msg_queue_destroy( struct object *obj ); static void msg_queue_poll_event( struct fd *fd, int event ); @@ -166,7 +165,7 @@ static const struct object_ops msg_queue_ops = &no_type, /* type */ msg_queue_dump, /* dump */ msg_queue_add_queue, /* add_queue */ - msg_queue_remove_queue, /* remove_queue */ + remove_queue, /* remove_queue */ msg_queue_signaled, /* signaled */ no_satisfied, /* satisfied */ no_signal, /* signal */ @@ -1307,33 +1306,10 @@ static int msg_queue_add_queue( struct object *obj, struct wait_queue_entry *ent return 0; } - if (queue->fd) - { - if (check_fd_events( queue->fd, POLLIN )) - set_queue_bits( queue, QS_DRIVER ); - else - { - clear_queue_bits( queue, QS_DRIVER ); - set_fd_events( queue->fd, POLLIN ); - } - } add_queue( obj, entry ); return 1; } -static void msg_queue_remove_queue(struct object *obj, struct wait_queue_entry *entry ) -{ - struct msg_queue *queue = (struct msg_queue *)obj; - - remove_queue( obj, entry ); - if (queue->fd) - { - /* after waiting, assume that all events will be processed */ - clear_queue_bits( queue, QS_DRIVER ); - set_fd_events( queue->fd, 0 ); - } -} - static void msg_queue_dump( struct object *obj, int verbose ) { struct msg_queue *queue = (struct msg_queue *)obj; @@ -3141,7 +3117,10 @@ DECL_HANDLER(set_queue_fd) if ((unix_fd = get_file_unix_fd( file )) != -1) { if ((unix_fd = dup( unix_fd )) != -1) + { queue->fd = create_anonymous_fd( &msg_queue_fd_ops, unix_fd, &queue->obj, 0 ); + set_fd_events( queue->fd, POLLIN ); + } else file_set_error(); } @@ -3161,8 +3140,8 @@ DECL_HANDLER(set_queue_mask) if (req->poll_events) { if (!queue->fd) return; - if (check_fd_events( queue->fd, POLLIN )) set_queue_bits( queue, QS_DRIVER ); - else clear_queue_bits( queue, QS_DRIVER ); + clear_queue_bits( queue, QS_DRIVER ); + set_fd_events( queue->fd, POLLIN ); return; } -- GitLab https://gitlab.winehq.org/wine/wine/-/merge_requests/9086
From: Rémi Bernon <rbernon(a)codeweavers.com> The thread queue handle is private and it shouldn't be possible to wait on it from other threads. --- server/queue.c | 18 +----------------- 1 file changed, 1 insertion(+), 17 deletions(-) diff --git a/server/queue.c b/server/queue.c index 7ae58bc2b2e..825ecb7217c 100644 --- a/server/queue.c +++ b/server/queue.c @@ -151,7 +151,6 @@ struct hotkey }; static void msg_queue_dump( struct object *obj, int verbose ); -static int msg_queue_add_queue( struct object *obj, struct wait_queue_entry *entry ); static int msg_queue_signaled( struct object *obj, struct wait_queue_entry *entry ); static void msg_queue_destroy( struct object *obj ); static void msg_queue_poll_event( struct fd *fd, int event ); @@ -164,7 +163,7 @@ static const struct object_ops msg_queue_ops = sizeof(struct msg_queue), /* size */ &no_type, /* type */ msg_queue_dump, /* dump */ - msg_queue_add_queue, /* add_queue */ + add_queue, /* add_queue */ remove_queue, /* remove_queue */ msg_queue_signaled, /* signaled */ no_satisfied, /* satisfied */ @@ -1295,21 +1294,6 @@ static int is_queue_hung( struct msg_queue *queue ) return queue->signaled && monotonic_time - queue->shared->access_time > 5 * TICKS_PER_SEC; } -static int msg_queue_add_queue( struct object *obj, struct wait_queue_entry *entry ) -{ - struct msg_queue *queue = (struct msg_queue *)obj; - - /* a thread can only wait on its own queue */ - if (get_wait_queue_thread(entry)->queue != queue) - { - set_error( STATUS_ACCESS_DENIED ); - return 0; - } - - add_queue( obj, entry ); - return 1; -} - static void msg_queue_dump( struct object *obj, int verbose ) { struct msg_queue *queue = (struct msg_queue *)obj; -- GitLab https://gitlab.winehq.org/wine/wine/-/merge_requests/9086
From: Rémi Bernon <rbernon(a)codeweavers.com> Instead of the dual inproc / server sync. --- server/inproc_sync.c | 5 +-- server/queue.c | 74 ++++++++++++++++---------------------------- server/user.h | 1 - 3 files changed, 27 insertions(+), 53 deletions(-) diff --git a/server/inproc_sync.c b/server/inproc_sync.c index 41a9c2770a7..435c58947f3 100644 --- a/server/inproc_sync.c +++ b/server/inproc_sync.c @@ -138,10 +138,7 @@ static int get_inproc_sync_fd( struct object *obj, int *type ) struct object *sync; int fd = -1; - if (obj != (struct object *)current->queue) sync = get_obj_sync( obj ); - else sync = thread_queue_inproc_sync( current ); - if (!sync) return -1; - + if (!(sync = get_obj_sync( obj ))) return -1; if (sync->ops == &inproc_sync_ops) { struct inproc_sync *inproc = (struct inproc_sync *)sync; diff --git a/server/queue.c b/server/queue.c index 825ecb7217c..97cb11c8cc0 100644 --- a/server/queue.c +++ b/server/queue.c @@ -119,8 +119,7 @@ struct msg_queue { struct object obj; /* object header */ struct fd *fd; /* optional file descriptor to poll */ - struct inproc_sync *inproc_sync; /* inproc sync for client-side waits */ - int signaled; /* queue is signaled from fd POLLIN or masks */ + struct event_sync *sync; /* sync object for wait/signal */ int paint_count; /* pending paint messages count */ int hotkey_count; /* pending hotkey messages count */ int quit_message; /* is there a pending quit message? */ @@ -151,7 +150,7 @@ struct hotkey }; static void msg_queue_dump( struct object *obj, int verbose ); -static int msg_queue_signaled( struct object *obj, struct wait_queue_entry *entry ); +static struct object *msg_queue_get_sync( struct object *obj ); static void msg_queue_destroy( struct object *obj ); static void msg_queue_poll_event( struct fd *fd, int event ); static void thread_input_dump( struct object *obj, int verbose ); @@ -163,13 +162,13 @@ static const struct object_ops msg_queue_ops = sizeof(struct msg_queue), /* size */ &no_type, /* type */ msg_queue_dump, /* dump */ - add_queue, /* add_queue */ - remove_queue, /* remove_queue */ - msg_queue_signaled, /* signaled */ - no_satisfied, /* satisfied */ + NULL, /* add_queue */ + NULL, /* remove_queue */ + NULL, /* signaled */ + NULL, /* satisfied */ no_signal, /* signal */ no_get_fd, /* get_fd */ - default_get_sync, /* get_sync */ + msg_queue_get_sync, /* get_sync */ default_map_access, /* map_access */ default_get_sd, /* get_sd */ default_set_sd, /* set_sd */ @@ -306,8 +305,7 @@ static struct msg_queue *create_msg_queue( struct thread *thread, struct thread_ if ((queue = alloc_object( &msg_queue_ops ))) { queue->fd = NULL; - queue->inproc_sync = NULL; - queue->signaled = 0; + queue->sync = NULL; queue->paint_count = 0; queue->hotkey_count = 0; queue->quit_message = 0; @@ -324,7 +322,7 @@ static struct msg_queue *create_msg_queue( struct thread *thread, struct thread_ list_init( &queue->expired_timers ); for (i = 0; i < NB_MSG_KINDS; i++) list_init( &queue->msg_list[i] ); - if (get_inproc_device_fd() >= 0 && !(queue->inproc_sync = create_inproc_internal_sync( 1, 0 ))) goto error; + if (!(queue->sync = create_event_sync( 1, 0 ))) goto error; if (!(queue->shared = alloc_shared_object())) goto error; SHARED_WRITE_BEGIN( queue->shared, queue_shm_t ) @@ -712,20 +710,6 @@ void add_queue_hook_count( struct thread *thread, unsigned int index, int count assert( thread->queue->shared->hooks_count[index] >= 0 ); } -static void signal_queue_sync( struct msg_queue *queue ) -{ - if (queue->signaled) return; - queue->signaled = 1; - wake_up( &queue->obj, 0 ); - if (queue->inproc_sync) signal_inproc_sync( queue->inproc_sync ); -} - -static void reset_queue_sync( struct msg_queue *queue ) -{ - queue->signaled = 0; - if (queue->inproc_sync) reset_inproc_sync( queue->inproc_sync ); -} - /* check the queue status */ static inline int get_queue_status( struct msg_queue *queue ) { @@ -756,7 +740,7 @@ static inline void set_queue_bits( struct msg_queue *queue, unsigned int bits ) } SHARED_WRITE_END; - if (get_queue_status( queue )) signal_queue_sync( queue ); + if (get_queue_status( queue )) signal_sync( queue->sync ); } /* clear some queue bits */ @@ -780,7 +764,7 @@ static inline void clear_queue_bits( struct msg_queue *queue, unsigned int bits if (queue->keystate_lock) unlock_input_keystate( queue->input ); queue->keystate_lock = 0; } - if (!get_queue_status( queue )) reset_queue_sync( queue ); + if (!get_queue_status( queue )) reset_sync( queue->sync ); } /* check if message is matched by the filter */ @@ -1291,22 +1275,22 @@ static void cleanup_results( struct msg_queue *queue ) static int is_queue_hung( struct msg_queue *queue ) { /* queue is hung if it's signaled and thread didn't access it for more than 5 seconds */ - return queue->signaled && monotonic_time - queue->shared->access_time > 5 * TICKS_PER_SEC; + return get_queue_status( queue ) && monotonic_time - queue->shared->access_time > 5 * TICKS_PER_SEC; } -static void msg_queue_dump( struct object *obj, int verbose ) +static struct object *msg_queue_get_sync( struct object *obj ) { struct msg_queue *queue = (struct msg_queue *)obj; - queue_shm_t *queue_shm = queue->shared; - fprintf( stderr, "Msg queue bits=%x mask=%x\n", - queue_shm->wake_bits, queue_shm->wake_mask ); + assert( obj->ops == &msg_queue_ops ); + return grab_object( queue->sync ); } -static int msg_queue_signaled( struct object *obj, struct wait_queue_entry *entry ) +static void msg_queue_dump( struct object *obj, int verbose ) { struct msg_queue *queue = (struct msg_queue *)obj; - assert( obj->ops == &msg_queue_ops ); - return queue->signaled; + queue_shm_t *queue_shm = queue->shared; + fprintf( stderr, "Msg queue bits=%x mask=%x\n", + queue_shm->wake_bits, queue_shm->wake_mask ); } static void msg_queue_destroy( struct object *obj ) @@ -1352,7 +1336,7 @@ static void msg_queue_destroy( struct object *obj ) if (queue->hooks) release_object( queue->hooks ); if (queue->fd) release_object( queue->fd ); if (queue->shared) free_shared_object( queue->shared ); - if (queue->inproc_sync) release_object( queue->inproc_sync ); + if (queue->sync) release_object( queue->sync ); } static void msg_queue_poll_event( struct fd *fd, int event ) @@ -1442,12 +1426,6 @@ int init_thread_queue( struct thread *thread ) return (create_msg_queue( thread, NULL ) != NULL); } -struct object *thread_queue_inproc_sync( struct thread *thread ) -{ - if (!thread->queue) return NULL; - return grab_object( thread->queue->inproc_sync ); -} - /* attach two thread input data structures */ int attach_thread_input( struct thread *thread_from, struct thread *thread_to ) { @@ -3139,8 +3117,8 @@ DECL_HANDLER(set_queue_mask) } SHARED_WRITE_END; - if (!get_queue_status( queue )) reset_queue_sync( queue ); - else signal_queue_sync( queue ); + if (!get_queue_status( queue )) reset_sync( queue->sync ); + else signal_sync( queue->sync ); } @@ -3161,7 +3139,7 @@ DECL_HANDLER(get_queue_status) } SHARED_WRITE_END; - if (!get_queue_status( queue )) reset_queue_sync( queue ); + if (!get_queue_status( queue )) reset_sync( queue->sync ); } @@ -3361,7 +3339,7 @@ DECL_HANDLER(get_message) } SHARED_WRITE_END; - if (!get_queue_status( queue )) reset_queue_sync( queue ); + if (!get_queue_status( queue )) reset_sync( queue->sync ); /* then check for posted messages */ if ((filter & QS_POSTMESSAGE) && @@ -3422,8 +3400,8 @@ DECL_HANDLER(get_message) } SHARED_WRITE_END; - if (!get_queue_status( queue )) reset_queue_sync( queue ); - else signal_queue_sync( queue ); + if (!get_queue_status( queue )) reset_sync( queue->sync ); + else signal_sync( queue->sync ); set_error( STATUS_PENDING ); /* FIXME */ } diff --git a/server/user.h b/server/user.h index 35bd396f859..73b10aaf5ac 100644 --- a/server/user.h +++ b/server/user.h @@ -120,7 +120,6 @@ extern void inc_queue_paint_count( struct thread *thread, int incr ); extern void queue_cleanup_window( struct thread *thread, user_handle_t win ); extern int init_thread_queue( struct thread *thread ); extern void check_thread_queue_idle( struct thread *thread ); -extern struct object *thread_queue_inproc_sync( struct thread *thread ); extern int attach_thread_input( struct thread *thread_from, struct thread *thread_to ); extern void detach_thread_input( struct thread *thread_from ); extern void set_clip_rectangle( struct desktop *desktop, const struct rectangle *rect, -- GitLab https://gitlab.winehq.org/wine/wine/-/merge_requests/9086
participants (1)
-
Rémi Bernon