server: Add support for associating a file descriptor to a message queue.
diff --git a/server/protocol.def b/server/protocol.def
index ef20235..48d1096 100644
--- a/server/protocol.def
+++ b/server/protocol.def
@@ -1783,6 +1783,12 @@
@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 */
diff --git a/server/queue.c b/server/queue.c
index 1973152..4e66f99 100644
--- a/server/queue.c
+++ b/server/queue.c
@@ -113,6 +113,7 @@
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 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 @@
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 @@
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 @@
{
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 @@
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 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 @@
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 @@
}
+/* 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)
{
diff --git a/server/request.h b/server/request.h
index aa4f8e4..5b28a7f 100644
--- a/server/request.h
+++ b/server/request.h
@@ -228,6 +228,7 @@
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 @@
(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,
diff --git a/server/trace.c b/server/trace.c
index c4572bd..0d11e7c 100644
--- a/server/trace.c
+++ b/server/trace.c
@@ -2214,6 +2214,11 @@
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 @@
(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 @@
(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 @@
"empty_atom_table",
"init_atom_table",
"get_msg_queue",
+ "set_queue_fd",
"set_queue_mask",
"get_queue_status",
"get_process_idle_event",
