server: Add support for associating a file descriptor to a message queue.

include/wine/server_protocol.h

@@ -2395,6 +2395,18 @@ struct get_msg_queue_reply
 
 
 
+struct set_queue_fd_request
+{
+    struct request_header __header;
+    obj_handle_t handle;
+};
+struct set_queue_fd_reply
+{
+    struct reply_header __header;
+};
+
+
+
 struct set_queue_mask_request
 {
     struct request_header __header;
@@ -4118,6 +4130,7 @@ enum request
     REQ_empty_atom_table,
     REQ_init_atom_table,
     REQ_get_msg_queue,
+    REQ_set_queue_fd,
     REQ_set_queue_mask,
     REQ_get_queue_status,
     REQ_get_process_idle_event,
@@ -4341,6 +4354,7 @@ union generic_request
     struct empty_atom_table_request empty_atom_table_request;
     struct init_atom_table_request init_atom_table_request;
     struct get_msg_queue_request get_msg_queue_request;
+    struct set_queue_fd_request set_queue_fd_request;
     struct set_queue_mask_request set_queue_mask_request;
     struct get_queue_status_request get_queue_status_request;
     struct get_process_idle_event_request get_process_idle_event_request;
@@ -4562,6 +4576,7 @@ union generic_reply
     struct empty_atom_table_reply empty_atom_table_reply;
     struct init_atom_table_reply init_atom_table_reply;
     struct get_msg_queue_reply get_msg_queue_reply;
+    struct set_queue_fd_reply set_queue_fd_reply;
     struct set_queue_mask_reply set_queue_mask_reply;
     struct get_queue_status_reply get_queue_status_reply;
     struct get_process_idle_event_reply get_process_idle_event_reply;
@@ -4662,6 +4677,6 @@ union generic_reply
     struct allocate_locally_unique_id_reply allocate_locally_unique_id_reply;
 };
 
-#define SERVER_PROTOCOL_VERSION 289
+#define SERVER_PROTOCOL_VERSION 290
 
 #endif /* __WINE_WINE_SERVER_PROTOCOL_H */

server/protocol.def

@@ -1783,6 +1783,12 @@ enum char_info_mode
 @END
 
 
+/* Set the file descriptor associated to the current thread queue */
+@REQ(set_queue_fd)
+    obj_handle_t handle;       /* handle to the file descriptor */
+@END
+
+
 /* Set the current message queue wakeup mask */
 @REQ(set_queue_mask)
     unsigned int wake_mask;    /* wakeup bits mask */

server/queue.c

@@ -113,6 +113,7 @@ struct thread_input
 struct msg_queue
 {
     struct object          obj;             /* object header */
+    struct fd             *fd;              /* optional file descriptor to poll */
     unsigned int           wake_bits;       /* wakeup bits */
     unsigned int           wake_mask;       /* wakeup mask */
     unsigned int           changed_bits;    /* changed wakeup bits */
@@ -139,6 +140,7 @@ static void msg_queue_remove_queue( struct object *obj, struct wait_queue_entry
 static int msg_queue_signaled( struct object *obj, struct thread *thread );
 static int msg_queue_satisfied( struct object *obj, struct thread *thread );
 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 );
 static void thread_input_destroy( struct object *obj );
 static void timer_callback( void *private );
@@ -160,6 +162,16 @@ static const struct object_ops msg_queue_ops =
     msg_queue_destroy          /* destroy */
 };
 
+static const struct fd_ops msg_queue_fd_ops =
+{
+    NULL,                        /* get_poll_events */
+    msg_queue_poll_event,        /* poll_event */
+    no_flush,                    /* flush */
+    no_get_file_info,            /* get_file_info */
+    no_queue_async,              /* queue_async */
+    no_cancel_async              /* cancel async */
+};
+
 
 static const struct object_ops thread_input_ops =
 {
@@ -243,6 +255,7 @@ static struct msg_queue *create_msg_queue( struct thread *thread, struct thread_
     if (!input && !(input = create_thread_input( thread ))) return NULL;
     if ((queue = alloc_object( &msg_queue_ops )))
     {
+        queue->fd              = NULL;
         queue->wake_bits       = 0;
         queue->wake_mask       = 0;
         queue->changed_bits    = 0;
@@ -778,6 +791,8 @@ static int msg_queue_add_queue( struct object *obj, struct wait_queue_entry *ent
     {
         if (process->idle_event) set_event( process->idle_event );
     }
+    if (queue->fd && list_empty( &obj->wait_queue ))  /* first on the queue */
+        set_fd_events( queue->fd, POLLIN );
     add_queue( obj, entry );
     return 1;
 }
@@ -788,6 +803,8 @@ static void msg_queue_remove_queue(struct object *obj, struct wait_queue_entry *
     struct process *process = entry->thread->process;
 
     remove_queue( obj, entry );
+    if (queue->fd && list_empty( &obj->wait_queue ))  /* last on the queue is gone */
+        set_fd_events( queue->fd, 0 );
 
     assert( entry->thread->queue == queue );
 
@@ -808,7 +825,18 @@ static void msg_queue_dump( struct object *obj, int verbose )
 static int msg_queue_signaled( struct object *obj, struct thread *thread )
 {
     struct msg_queue *queue = (struct msg_queue *)obj;
-    return is_signaled( queue );
+    int ret = 0;
+
+    if (queue->fd)
+    {
+        if ((ret = check_fd_events( queue->fd, POLLIN )))
+            /* stop waiting on select() if we are signaled */
+            set_fd_events( queue->fd, 0 );
+        else if (!list_empty( &obj->wait_queue ))
+            /* restart waiting on poll() if we are no longer signaled */
+            set_fd_events( queue->fd, POLLIN );
+    }
+    return ret || is_signaled( queue );
 }
 
 static int msg_queue_satisfied( struct object *obj, struct thread *thread )
@@ -843,6 +871,16 @@ static void msg_queue_destroy( struct object *obj )
     if (queue->timeout) remove_timeout_user( queue->timeout );
     if (queue->input) release_object( queue->input );
     if (queue->hooks) release_object( queue->hooks );
+    if (queue->fd) release_object( queue->fd );
+}
+
+static void msg_queue_poll_event( struct fd *fd, int event )
+{
+    struct msg_queue *queue = get_fd_user( fd );
+    assert( queue->obj.ops == &msg_queue_ops );
+
+    if (event & (POLLERR | POLLHUP)) set_fd_events( fd, -1 );
+    wake_up( &queue->obj, 0 );
 }
 
 static void thread_input_dump( struct object *obj, int verbose )
@@ -1524,6 +1562,31 @@ DECL_HANDLER(get_msg_queue)
 }
 
 
+/* set the file descriptor associated to the current thread queue */
+DECL_HANDLER(set_queue_fd)
+{
+    struct msg_queue *queue = get_current_queue();
+    struct file *file;
+    int unix_fd;
+
+    if (queue->fd)  /* fd can only be set once */
+    {
+        set_error( STATUS_ACCESS_DENIED );
+        return;
+    }
+    if (!(file = get_file_obj( current->process, req->handle, SYNCHRONIZE ))) return;
+
+    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 );
+        else
+            file_set_error();
+    }
+    release_object( file );
+}
+
+
 /* set the current message queue wakeup mask */
 DECL_HANDLER(set_queue_mask)
 {

server/request.h

@@ -228,6 +228,7 @@ DECL_HANDLER(set_atom_information);
 DECL_HANDLER(empty_atom_table);
 DECL_HANDLER(init_atom_table);
 DECL_HANDLER(get_msg_queue);
+DECL_HANDLER(set_queue_fd);
 DECL_HANDLER(set_queue_mask);
 DECL_HANDLER(get_queue_status);
 DECL_HANDLER(get_process_idle_event);
@@ -450,6 +451,7 @@ static const req_handler req_handlers[REQ_NB_REQUESTS] =
     (req_handler)req_empty_atom_table,
     (req_handler)req_init_atom_table,
     (req_handler)req_get_msg_queue,
+    (req_handler)req_set_queue_fd,
     (req_handler)req_set_queue_mask,
     (req_handler)req_get_queue_status,
     (req_handler)req_get_process_idle_event,

server/trace.c

@@ -2214,6 +2214,11 @@ static void dump_get_msg_queue_reply( const struct get_msg_queue_reply *req )
     fprintf( stderr, " handle=%p", req->handle );
 }
 
+static void dump_set_queue_fd_request( const struct set_queue_fd_request *req )
+{
+    fprintf( stderr, " handle=%p", req->handle );
+}
+
 static void dump_set_queue_mask_request( const struct set_queue_mask_request *req )
 {
     fprintf( stderr, " wake_mask=%08x,", req->wake_mask );
@@ -3568,6 +3573,7 @@ static const dump_func req_dumpers[REQ_NB_REQUESTS] = {
     (dump_func)dump_empty_atom_table_request,
     (dump_func)dump_init_atom_table_request,
     (dump_func)dump_get_msg_queue_request,
+    (dump_func)dump_set_queue_fd_request,
     (dump_func)dump_set_queue_mask_request,
     (dump_func)dump_get_queue_status_request,
     (dump_func)dump_get_process_idle_event_request,
@@ -3787,6 +3793,7 @@ static const dump_func reply_dumpers[REQ_NB_REQUESTS] = {
     (dump_func)0,
     (dump_func)dump_init_atom_table_reply,
     (dump_func)dump_get_msg_queue_reply,
+    (dump_func)0,
     (dump_func)dump_set_queue_mask_reply,
     (dump_func)dump_get_queue_status_reply,
     (dump_func)dump_get_process_idle_event_reply,
@@ -4006,6 +4013,7 @@ static const char * const req_names[REQ_NB_REQUESTS] = {
     "empty_atom_table",
     "init_atom_table",
     "get_msg_queue",
+    "set_queue_fd",
     "set_queue_mask",
     "get_queue_status",
     "get_process_idle_event",