server: Add support for associating a file descriptor to a message queue.
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)
{
