Added add_queue/remove_queue to server object operations.

include/server/object.h

@@ -20,13 +20,22 @@
 struct object;
 struct object_name;
 struct thread;
+struct wait_queue_entry;
 
 struct object_ops
 {
-    void (*dump)(struct object *,int);                  /* dump the object (for debugging) */
-    int  (*signaled)(struct object *,struct thread *);  /* is object signaled? */
-    int  (*satisfied)(struct object *,struct thread *); /* wait satisfied; return 1 if abandoned */
-    void (*destroy)(struct object *);                   /* destroy on refcount == 0 */
+    /* dump the object (for debugging) */
+    void (*dump)(struct object *,int);
+    /* add a thread to the object wait queue */
+    void (*add_queue)(struct object *,struct wait_queue_entry *);
+    /* remove a thread from the object wait queue */
+    void (*remove_queue)(struct object *,struct wait_queue_entry *);
+    /* is object signaled? */
+    int  (*signaled)(struct object *,struct thread *);
+    /* wait satisfied; return 1 if abandoned */
+    int  (*satisfied)(struct object *,struct thread *);
+    /* destroy on refcount == 0 */
+    void (*destroy)(struct object *);
 };
 
 struct object
@@ -65,8 +74,27 @@ extern void trace_kill( int exit_code );
 extern void trace_reply( struct thread *thread, int type, int pass_fd,
                          struct iovec *vec, int veclen );
 
+/* select functions */
+
+#define READ_EVENT    1
+#define WRITE_EVENT   2
+
+struct select_ops
+{
+    void (*event)( int fd, int event, void *private );
+    void (*timeout)( int fd, void *private );
+};
+
+extern int add_select_user( int fd, int events, const struct select_ops *ops, void *private );
+extern void remove_select_user( int fd );
+extern void set_select_timeout( int fd, struct timeval *when );
+extern void set_select_events( int fd, int events );
+extern void *get_select_private_data( const struct select_ops *ops, int fd );
+extern void select_loop(void);
+
 /* socket functions */
 
+extern void server_init( int fd );
 extern int add_client( int client_fd, struct thread *self );
 extern void remove_client( int client_fd, int exit_code );
 extern int get_initial_client_fd(void);

include/server/thread.h

@@ -52,6 +52,8 @@ extern void get_thread_info( struct thread *thread,
                              struct get_thread_info_reply *reply );
 extern int send_reply( struct thread *thread, int pass_fd,
                        int n, ... /* arg_1, len_1, ..., arg_n, len_n */ );
+extern void 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 exit_code );
 extern void thread_killed( struct thread *thread, int exit_code );
 extern void thread_timeout(void);

scheduler/client.c

@@ -226,7 +226,8 @@ int CLIENT_NewThread( THDB *thdb, int *thandle, int *phandle )
     struct new_thread_reply reply;
     int len, fd[2];
     extern BOOL32 THREAD_InitDone;
-    extern void server_main_loop( int fd );
+    extern void server_init( int fd );
+    extern void select_loop(void);
 
     if (!THREAD_InitDone)  /* first thread -> start the server */
     {
@@ -251,7 +252,8 @@ int CLIENT_NewThread( THDB *thdb, int *thandle, int *phandle )
             execl( "/usr/local/bin/wineserver", "wineserver", buffer, NULL );
             execl( "./server/wineserver", "wineserver", buffer, NULL );
 #endif
-            server_main_loop( tmpfd[1] );
+            server_init( tmpfd[1] );
+            select_loop();
             exit(0);
         default:  /* parent */
             close( tmpfd[1] );

server/Makefile.in

@@ -11,6 +11,7 @@ C_SRCS = \
 	object.c \
 	process.c \
 	request.c \
+	select.c \
 	semaphore.c \
 	socket.c \
 	thread.c \

server/event.c

@@ -27,6 +27,8 @@ static void destroy_event( struct object *obj );
 static const struct object_ops event_ops =
 {
     dump_event,
+    add_queue,
+    remove_queue,
     event_signaled,
     event_satisfied,
     destroy_event

server/main.c

@@ -16,7 +16,6 @@
 int main( int argc, char *argv[] )
 {
     int fd;
-    extern void server_main_loop( int fd );
 
     if (argc != 2) goto error;
     if (!isdigit( *argv[1] )) goto error;
@@ -27,7 +26,8 @@ int main( int argc, char *argv[] )
     debug_level = 1;
 
     if (debug_level) printf( "Server: starting (pid=%d)\n", getpid() );
-    server_main_loop( fd );
+    server_init( fd );
+    select_loop();
     if (debug_level) printf( "Server: exiting (pid=%d)\n", getpid() );
     exit(0);
 

server/mutex.c

@@ -30,6 +30,8 @@ static void destroy_mutex( struct object *obj );
 static const struct object_ops mutex_ops =
 {
     dump_mutex,
+    add_queue,
+    remove_queue,
     mutex_signaled,
     mutex_satisfied,
     destroy_mutex

server/process.c

@@ -64,6 +64,8 @@ static int copy_handle_table( struct process *process, struct process *parent );
 static const struct object_ops process_ops =
 {
     dump_process,
+    add_queue,
+    remove_queue,
     process_signaled,
     process_satisfied,
     destroy_process

server/select.c

+/*
+ * Server main select() loop
+ *
+ * Copyright (C) 1998 Alexandre Julliard
+ */
+
+#include <assert.h>
+#include <errno.h>
+#include <fcntl.h>
+#include <signal.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/time.h>
+#include <sys/types.h>
+#include <unistd.h>
+
+#include "server/object.h"
+
+/* select user fd */
+struct user
+{
+    struct timeval           when;    /* timeout expiry (absolute time) */
+    struct user             *next;    /* next in sorted timeout list */
+    struct user             *prev;    /* prev in sorted timeout list */
+    const struct select_ops *ops;   /* user operations list */
+    int                      fd;      /* user fd */
+    void                    *private; /* user private data */
+};
+
+static struct user *users[FD_SETSIZE];      /* users array */
+static fd_set read_set, write_set;          /* current select sets */
+static int nb_users;                        /* current number of users */
+static int max_fd;                          /* max fd in use */
+static struct user *timeout_head;           /* sorted timeouts list head */
+static struct user *timeout_tail;           /* sorted timeouts list tail */
+
+
+/* add a user */
+int add_select_user( int fd, int events, const struct select_ops *ops, void *private )
+{
+    int flags;
+    struct user *user = malloc( sizeof(*user) );
+    if (!user) return -1;
+    assert( !users[fd] );
+
+    user->ops          = ops;
+    user->when.tv_sec  = 0;
+    user->when.tv_usec = 0;
+    user->fd           = fd;
+    user->private      = private;
+
+    flags = fcntl( fd, F_GETFL, 0 );
+    fcntl( fd, F_SETFL, flags | O_NONBLOCK );
+
+    users[fd] = user;
+    set_select_events( fd, events );
+    if (fd > max_fd) max_fd = fd;
+    nb_users++;
+    return fd;
+}
+
+/* remove a user */
+void remove_select_user( int fd )
+{
+    struct user *user = users[fd];
+    assert( user );
+
+    set_select_timeout( fd, 0 );
+    set_select_events( fd, 0 );
+    users[fd] = NULL;
+    if (max_fd == fd) while (max_fd && !users[max_fd]) max_fd--;
+    nb_users--;
+    free( user );
+}
+
+/* set a user timeout */
+void set_select_timeout( int fd, struct timeval *when )
+{
+    struct user *user = users[fd];
+    struct user *pos;
+    assert( user );
+
+    if (user->when.tv_sec || user->when.tv_usec)
+    {
+        /* there is already a timeout */
+        if (user->next) user->next->prev = user->prev;
+        else timeout_tail = user->prev;
+        if (user->prev) user->prev->next = user->next;
+        else timeout_head = user->next;
+        user->when.tv_sec = user->when.tv_usec = 0;
+    }
+    if (!when) return;  /* no timeout */
+    user->when = *when;
+
+    /* Now insert it in the linked list */
+
+    for (pos = timeout_head; pos; pos = pos->next)
+    {
+        if (pos->when.tv_sec > user->when.tv_sec) break;
+        if ((pos->when.tv_sec == user->when.tv_sec) &&
+            (pos->when.tv_usec > user->when.tv_usec)) break;
+    }
+
+    if (pos)  /* insert it before 'pos' */
+    {
+        if ((user->prev = pos->prev)) user->prev->next = user;
+        else timeout_head = user;
+        user->next = pos;
+        pos->prev = user;
+    }
+    else  /* insert it at the tail */
+    {
+        user->next = NULL;
+        if (timeout_tail) timeout_tail->next = user;
+        else timeout_head = user;
+        user->prev = timeout_tail;
+        timeout_tail = user;
+    }
+}
+
+/* set the events that select waits for on this fd */
+void set_select_events( int fd, int events )
+{
+    if (events & READ_EVENT) FD_SET( fd, &read_set );
+    else FD_CLR( fd, &read_set );
+    if (events & WRITE_EVENT) FD_SET( fd, &write_set );
+    else FD_CLR( fd, &write_set );
+}
+
+/* get a user private data, checking the type */
+void *get_select_private_data( const struct select_ops *ops, int fd )
+{
+    struct user *user = users[fd];
+    assert( user );
+    assert( user->ops == ops );
+    return user->private;
+}
+
+/* server main loop */
+void select_loop(void)
+{
+    int i, ret;
+
+    setsid();
+    signal( SIGPIPE, SIG_IGN );
+
+    while (nb_users)
+    {
+        fd_set read = read_set, write = write_set;
+#if 0
+        printf( "select: " );
+        for (i = 0; i <= max_fd; i++) printf( "%c", FD_ISSET( i, &read_set ) ? 'r' :
+                                                    (FD_ISSET( i, &write_set ) ? 'w' : '-') );
+        printf( "\n" );
+#endif
+        if (timeout_head)
+        {
+            struct timeval tv, now;
+            gettimeofday( &now, NULL );
+            if ((timeout_head->when.tv_sec < now.tv_sec) ||
+                ((timeout_head->when.tv_sec == now.tv_sec) &&
+                 (timeout_head->when.tv_usec < now.tv_usec)))
+            {
+                timeout_head->ops->timeout( timeout_head->fd, timeout_head->private );
+                continue;
+            }
+            tv.tv_sec = timeout_head->when.tv_sec - now.tv_sec;
+            if ((tv.tv_usec = timeout_head->when.tv_usec - now.tv_usec) < 0)
+            {
+                tv.tv_usec += 1000000;
+                tv.tv_sec--;
+            }
+            ret = select( max_fd + 1, &read, &write, NULL, &tv );
+        }
+        else  /* no timeout */
+        {
+            ret = select( max_fd + 1, &read, &write, NULL, NULL );
+        }
+
+        if (!ret) continue;
+        if (ret == -1) perror("select");
+
+        for (i = 0; i <= max_fd; i++)
+        {
+            int event = 0;
+            if (FD_ISSET( i, &write )) event |= WRITE_EVENT;
+            if (FD_ISSET( i, &read ))  event |= READ_EVENT;
+            if (event) users[i]->ops->event( i, event, users[i]->private );
+        }
+    }
+}

server/semaphore.c

@@ -27,6 +27,8 @@ static void destroy_semaphore( struct object *obj );
 static const struct object_ops semaphore_ops =
 {
     dump_semaphore,
+    add_queue,
+    remove_queue,
     semaphore_signaled,
     semaphore_satisfied,
     destroy_semaphore

server/socket.c

@@ -6,8 +6,6 @@
 
 #include <assert.h>
 #include <errno.h>
-#include <fcntl.h>
-#include <signal.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <stdarg.h>
@@ -37,15 +35,6 @@ enum state
     READING    /* reading our reply */
 };
 
-/* client timeout */
-struct timeout
-{
-    struct timeval  when;    /* timeout expiry (absolute time) */
-    struct timeout *next;    /* next in sorted list */
-    struct timeout *prev;    /* prev in sorted list */
-    int             client;  /* client id */
-};
-
 /* client structure */
 struct client
 {
@@ -56,17 +45,9 @@ struct client
     int                count;        /* bytes sent/received so far */
     int                pass_fd;      /* fd to pass to and from the client */
     struct thread     *self;         /* client thread (opaque pointer) */
-    struct timeout     timeout;      /* client timeout */
 };
 
-
-static struct client *clients[FD_SETSIZE];  /* clients array */
-static fd_set read_set, write_set;          /* current select sets */
-static int nb_clients;                      /* current number of clients */
-static int max_fd;                          /* max fd in use */
 static int initial_client_fd;               /* fd of the first client */
-static struct timeout *timeout_head;        /* sorted timeouts list head */
-static struct timeout *timeout_tail;        /* sorted timeouts list tail */
 
 /* exit code passed to remove_client */
 #define OUT_OF_MEMORY  -1
@@ -85,11 +66,9 @@ static void protocol_error( int client_fd, const char *err, ... )
     va_end( args );
 }
 
-
 /* send a message to a client that is ready to receive something */
-static void do_write( int client_fd )
+static void do_write( struct client *client, int client_fd )
 {
-    struct client *client = clients[client_fd];
     struct iovec vec[2];
 #ifndef HAVE_MSGHDR_ACCRIGHTS
     struct cmsg_fd cmsg  = { sizeof(cmsg), SOL_SOCKET, SCM_RIGHTS,
@@ -146,15 +125,13 @@ static void do_write( int client_fd )
     client->count = 0;
     client->state = RUNNING;
     client->seq++;
-    FD_CLR( client_fd, &write_set );
-    FD_SET( client_fd, &read_set );
+    set_select_events( client_fd, READ_EVENT );
 }
 
 
 /* read a message from a client that has something to say */
-static void do_read( int client_fd )
+static void do_read( struct client *client, int client_fd )
 {
-    struct client *client = clients[client_fd];
     struct iovec vec;
     int pass_fd = -1;
 #ifdef HAVE_MSGHDR_ACCRIGHTS
@@ -254,89 +231,47 @@ static void do_read( int client_fd )
     }
 }
 
-
 /* handle a client timeout */
-static void do_timeout( int client_fd )
+static void client_timeout( int client_fd, void *private )
 {
-    struct client *client = clients[client_fd];
-    set_timeout( client_fd, 0 );  /* Remove the timeout */
+    struct client *client = (struct client *)private;
+    set_select_timeout( client_fd, 0 );  /* Remove the timeout */
     call_timeout_handler( client->self );
 }
 
+/* handle a client event */
+static void client_event( int client_fd, int event, void *private )
+{
+    struct client *client = (struct client *)private;
+    if (event & WRITE_EVENT)
+        do_write( client, client_fd );
+    if (event & READ_EVENT)
+        do_read( client, client_fd );
+}
 
-/* server main loop */
-void server_main_loop( int fd )
+static const struct select_ops client_ops =
 {
-    int i, ret;
+    client_event,
+    client_timeout
+};
 
-    setsid();
-    signal( SIGPIPE, SIG_IGN );
+/*******************************************************************/
+/* server-side exported functions                                  */
 
+/* server initialization */
+void server_init( int fd )
+{
     /* special magic to create the initial thread */
     initial_client_fd = fd;
     add_client( initial_client_fd, NULL );
-
-    while (nb_clients)
-    {
-        fd_set read = read_set, write = write_set;
-#if 0
-        printf( "select: " );
-        for (i = 0; i <= max_fd; i++) printf( "%c", FD_ISSET( i, &read_set ) ? 'r' :
-                                                    (FD_ISSET( i, &write_set ) ? 'w' : '-') );
-        printf( "\n" );
-#endif
-        if (timeout_head)
-        {
-            struct timeval tv, now;
-            gettimeofday( &now, NULL );
-            if ((timeout_head->when.tv_sec < now.tv_sec) ||
-                ((timeout_head->when.tv_sec == now.tv_sec) &&
-                 (timeout_head->when.tv_usec < now.tv_usec)))
-            {
-                do_timeout( timeout_head->client );
-                continue;
-            }
-            tv.tv_sec = timeout_head->when.tv_sec - now.tv_sec;
-            if ((tv.tv_usec = timeout_head->when.tv_usec - now.tv_usec) < 0)
-            {
-                tv.tv_usec += 1000000;
-                tv.tv_sec--;
-            }
-            ret = select( max_fd + 1, &read, &write, NULL, &tv );
-        }
-        else  /* no timeout */
-        {
-            ret = select( max_fd + 1, &read, &write, NULL, NULL );
-        }
-
-        if (!ret) continue;
-        if (ret == -1) perror("select");
-
-        for (i = 0; i <= max_fd; i++)
-        {
-            if (FD_ISSET( i, &write ))
-            {
-                if (clients[i]) do_write( i );
-            }
-            else if (FD_ISSET( i, &read ))
-            {
-                if (clients[i]) do_read( i );
-            }
-        }
-    }
 }
 
 
-/*******************************************************************/
-/* server-side exported functions                                  */
-
 /* add a client */
 int add_client( int client_fd, struct thread *self )
 {
-    int flags;
     struct client *client = malloc( sizeof(*client) );
     if (!client) return -1;
-    assert( !clients[client_fd] );
 
     client->state                = RUNNING;
     client->seq                  = 0;
@@ -346,36 +281,26 @@ int add_client( int client_fd, struct thread *self )
     client->data                 = NULL;
     client->self                 = self;
     client->pass_fd              = -1;
-    client->timeout.when.tv_sec  = 0;
-    client->timeout.when.tv_usec = 0;
-    client->timeout.client       = client_fd;
 
-    flags = fcntl( client_fd, F_GETFL, 0 );
-    fcntl( client_fd, F_SETFL, flags | O_NONBLOCK );
-
-    clients[client_fd] = client;
-    FD_SET( client_fd, &read_set );
-    if (client_fd > max_fd) max_fd = client_fd;
-    nb_clients++;
+    if (add_select_user( client_fd, READ_EVENT, &client_ops, client ) == -1)
+    {
+        free( client );
+        return -1;
+    }
     return client_fd;
 }
 
 /* remove a client */
 void remove_client( int client_fd, int exit_code )
 {
-    struct client *client = clients[client_fd];
+    struct client *client = (struct client *)get_select_private_data( &client_ops, client_fd );
     assert( client );
 
     call_kill_handler( client->self, exit_code );
 
-    set_timeout( client_fd, 0 );
-    clients[client_fd] = NULL;
-    FD_CLR( client_fd, &read_set );
-    FD_CLR( client_fd, &write_set );
-    if (max_fd == client_fd) while (max_fd && !clients[max_fd]) max_fd--;
+    remove_select_user( client_fd );
     if (initial_client_fd == client_fd) initial_client_fd = -1;
     close( client_fd );
-    nb_clients--;
 
     /* Purge messages */
     if (client->data) free( client->data );
@@ -390,53 +315,6 @@ int get_initial_client_fd(void)
     return initial_client_fd;
 }
 
-/* set a client timeout */
-void set_timeout( int client_fd, struct timeval *when )
-{
-    struct timeout *tm, *pos;
-    struct client *client = clients[client_fd];
-    assert( client );
-
-    tm = &client->timeout;
-    if (tm->when.tv_sec || tm->when.tv_usec)
-    {
-        /* there is already a timeout */
-        if (tm->next) tm->next->prev = tm->prev;
-        else timeout_tail = tm->prev;
-        if (tm->prev) tm->prev->next = tm->next;
-        else timeout_head = tm->next;
-        tm->when.tv_sec = tm->when.tv_usec = 0;
-    }
-    if (!when) return;  /* no timeout */
-    tm->when = *when;
-
-    /* Now insert it in the linked list */
-
-    for (pos = timeout_head; pos; pos = pos->next)
-    {
-        if (pos->when.tv_sec > tm->when.tv_sec) break;
-        if ((pos->when.tv_sec == tm->when.tv_sec) &&
-            (pos->when.tv_usec > tm->when.tv_usec)) break;
-    }
-
-    if (pos)  /* insert it before 'pos' */
-    {
-        if ((tm->prev = pos->prev)) tm->prev->next = tm;
-        else timeout_head = tm;
-        tm->next = pos;
-        pos->prev = tm;
-    }
-    else  /* insert it at the tail */
-    {
-        tm->next = NULL;
-        if (timeout_tail) timeout_tail->next = tm;
-        else timeout_head = tm;
-        tm->prev = timeout_tail;
-        timeout_tail = tm;
-    }
-}
-
-
 /* send a reply to a client */
 int send_reply_v( int client_fd, int type, int pass_fd,
                   struct iovec *vec, int veclen )
@@ -444,7 +322,7 @@ int send_reply_v( int client_fd, int type, int pass_fd,
     int i;
     unsigned int len;
     char *p;
-    struct client *client = clients[client_fd];
+    struct client *client = (struct client *)get_select_private_data( &client_ops, client_fd );
 
     assert( client );
     assert( client->state == WAITING );
@@ -469,7 +347,6 @@ int send_reply_v( int client_fd, int type, int pass_fd,
     }
 
     client->state = READING;
-    FD_CLR( client_fd, &read_set );
-    FD_SET( client_fd, &write_set );
+    set_select_events( client_fd, WRITE_EVENT );
     return 0;
 }

server/thread.c

@@ -49,6 +49,8 @@ static void destroy_thread( struct object *obj );
 static const struct object_ops thread_ops =
 {
     dump_thread,
+    add_queue,
+    remove_queue,
     thread_signaled,
     thread_satisfied,
     destroy_thread
@@ -200,8 +202,9 @@ int send_reply( struct thread *thread, int pass_fd, int n,
 }
 
 /* add a thread to an object wait queue; return 1 if OK, 0 on error */
-static void add_queue( struct object *obj, struct wait_queue_entry *entry )
+void add_queue( struct object *obj, struct wait_queue_entry *entry )
 {
+    grab_object( obj );
     entry->obj    = obj;
     entry->prev   = obj->tail;
     entry->next   = NULL;
@@ -211,10 +214,8 @@ static void add_queue( struct object *obj, struct wait_queue_entry *entry )
 }
 
 /* remove a thread from an object wait queue */
-static void remove_queue( struct wait_queue_entry *entry )
+void remove_queue( struct object *obj, struct wait_queue_entry *entry )
 {
-    struct object *obj = entry->obj;
-
     if (entry->next) entry->next->prev = entry->prev;
     else obj->tail = entry->prev;
     if (entry->prev) entry->prev->next = entry->next;
@@ -230,9 +231,9 @@ static void end_wait( struct thread *thread )
     int i;
 
     assert( wait );
-    for (i = 0, entry = wait->queues; i < wait->count; i++)
-        remove_queue( entry++ );
-    if (wait->flags & SELECT_TIMEOUT) set_timeout( thread->client_fd, NULL );
+    for (i = 0, entry = wait->queues; i < wait->count; i++, entry++)
+        entry->obj->ops->remove_queue( entry->obj, entry );
+    if (wait->flags & SELECT_TIMEOUT) set_select_timeout( thread->client_fd, NULL );
     free( wait );
     thread->wait = NULL;
 }
@@ -280,7 +281,8 @@ static int wait_on( struct thread *thread, int count,
             return 0;
         }
         entry->thread = thread;
-        add_queue( obj, entry );
+        obj->ops->add_queue( obj, entry );
+        release_object( obj );
     }
     return 1;
 }
@@ -342,7 +344,7 @@ void sleep_on( struct thread *thread, int count, int *handles, int flags, int ti
     {
         /* we need to wait */
         if (flags & SELECT_TIMEOUT)
-            set_timeout( thread->client_fd, &thread->wait->timeout );
+            set_select_timeout( thread->client_fd, &thread->wait->timeout );
         return;
     }
     end_wait( thread );