server/thread.c

   1 /*
   2  * Server-side thread management
   3  *
   4  * Copyright (C) 1998 Alexandre Julliard
   5  */
   6
   7 #include <assert.h>
   8 #include <fcntl.h>
   9 #include <signal.h>
  10 #include <stdio.h>
  11 #include <stdlib.h>
  12 #include <string.h>
  13 #include <sys/types.h>
  14 #include <sys/uio.h>
  15 #include <unistd.h>
  16
  17 #include "winnt.h"
  18 #include "winerror.h"
  19 #include "server.h"
  20 #include "server/thread.h"
  21
  22
  23 /* thread queues */
  24
  25 struct wait_queue_entry
  26 {
  27     struct wait_queue_entry *next;
  28     struct wait_queue_entry *prev;
  29     struct object           *obj;
  30     struct thread           *thread;
  31 };
  32
  33 struct thread_wait
  34 {
  35     int                     count;      /* count of objects */
  36     int                     flags;
  37     struct timeval          timeout;
  38     struct wait_queue_entry queues[1];
  39 };
  40
  41
  42 /* thread operations */
  43
  44 static void dump_thread( struct object *obj, int verbose );
  45 static int thread_signaled( struct object *obj, struct thread *thread );
  46 static void destroy_thread( struct object *obj );
  47
  48 static const struct object_ops thread_ops =
  49 {
  50     dump_thread,
  51     add_queue,
  52     remove_queue,
  53     thread_signaled,
  54     no_satisfied,
  55     no_read_fd,
  56     no_write_fd,
  57     no_flush,
  58     destroy_thread
  59 };
  60
  61 static struct thread *first_thread;
  62
  63
  64 /* create a new thread */
  65 struct thread *create_thread( int fd, void *pid, int *thread_handle,
  66                               int *process_handle )
  67 {
  68     struct thread *thread;
  69     struct process *process;
  70
  71     if (!(thread = mem_alloc( sizeof(*thread) ))) return NULL;
  72
  73     if (pid) process = get_process_from_id( pid );
  74     else process = create_process();
  75     if (!process)
  76     {
  77         free( thread );
  78         return NULL;
  79     }
  80
  81     init_object( &thread->obj, &thread_ops, NULL );
  82     thread->client_fd = fd;
  83     thread->process   = process;
  84     thread->unix_pid  = 0;  /* not known yet */
  85     thread->name      = NULL;
  86     thread->mutex     = NULL;
  87     thread->wait      = NULL;
  88     thread->error     = 0;
  89     thread->state     = STARTING;
  90     thread->exit_code = 0x103;  /* STILL_ACTIVE */
  91     thread->next      = first_thread;
  92     thread->prev      = NULL;
  93
  94     if (first_thread) first_thread->prev = thread;
  95     first_thread = thread;
  96     add_process_thread( process, thread );
  97
  98     *thread_handle = *process_handle = -1;
  99     if (current)
 100     {
 101         if ((*thread_handle = alloc_handle( current->process, thread,
 102                                             THREAD_ALL_ACCESS, 0 )) == -1)
 103             goto error;
 104     }
 105     if (current && !pid)
 106     {
 107         if ((*process_handle = alloc_handle( current->process, process,
 108                                              PROCESS_ALL_ACCESS, 0 )) == -1)
 109             goto error;
 110     }
 111
 112     if (add_client( fd, thread ) == -1) goto error;
 113
 114     return thread;
 115
 116  error:
 117     if (current)
 118     {
 119         close_handle( current->process, *thread_handle );
 120         close_handle( current->process, *process_handle );
 121     }
 122     remove_process_thread( process, thread );
 123     release_object( thread );
 124     return NULL;
 125 }
 126
 127 /* destroy a thread when its refcount is 0 */
 128 static void destroy_thread( struct object *obj )
 129 {
 130     struct thread *thread = (struct thread *)obj;
 131     assert( obj->ops == &thread_ops );
 132
 133     release_object( thread->process );
 134     if (thread->next) thread->next->prev = thread->prev;
 135     if (thread->prev) thread->prev->next = thread->next;
 136     else first_thread = thread->next;
 137     if (thread->name) free( thread->name );
 138     if (debug_level) memset( thread, 0xaa, sizeof(thread) );  /* catch errors */
 139     free( thread );
 140 }
 141
 142 /* dump a thread on stdout for debugging purposes */
 143 static void dump_thread( struct object *obj, int verbose )
 144 {
 145     struct thread *thread = (struct thread *)obj;
 146     assert( obj->ops == &thread_ops );
 147
 148     printf( "Thread pid=%d fd=%d name='%s'\n",
 149             thread->unix_pid, thread->client_fd, thread->name );
 150 }
 151
 152 static int thread_signaled( struct object *obj, struct thread *thread )
 153 {
 154     struct thread *mythread = (struct thread *)obj;
 155     return (mythread->state == TERMINATED);
 156 }
 157
 158 /* get a thread pointer from a thread id (and increment the refcount) */
 159 struct thread *get_thread_from_id( void *id )
 160 {
 161     struct thread *t = first_thread;
 162     while (t && (t != id)) t = t->next;
 163     if (t) grab_object( t );
 164     return t;
 165 }
 166
 167 /* get a thread from a handle (and increment the refcount) */
 168 struct thread *get_thread_from_handle( int handle, unsigned int access )
 169 {
 170     return (struct thread *)get_handle_obj( current->process, handle,
 171                                             access, &thread_ops );
 172 }
 173
 174 /* get all information about a thread */
 175 void get_thread_info( struct thread *thread,
 176                       struct get_thread_info_reply *reply )
 177 {
 178     reply->pid       = thread;
 179     reply->exit_code = thread->exit_code;
 180 }
 181
 182 /* send a reply to a thread */
 183 int send_reply( struct thread *thread, int pass_fd, int n,
 184                 ... /* arg_1, len_1, ..., arg_n, len_n */ )
 185 {
 186     struct iovec vec[16];
 187     va_list args;
 188     int i;
 189
 190     assert( n < 16 );
 191     va_start( args, n );
 192     for (i = 0; i < n; i++)
 193     {
 194         vec[i].iov_base = va_arg( args, void * );
 195         vec[i].iov_len  = va_arg( args, int );
 196     }
 197     va_end( args );
 198     return send_reply_v( thread->client_fd, thread->error, pass_fd, vec, n );
 199 }
 200
 201 /* add a thread to an object wait queue; return 1 if OK, 0 on error */
 202 void add_queue( struct object *obj, struct wait_queue_entry *entry )
 203 {
 204     grab_object( obj );
 205     entry->obj    = obj;
 206     entry->prev   = obj->tail;
 207     entry->next   = NULL;
 208     if (obj->tail) obj->tail->next = entry;
 209     else obj->head = entry;
 210     obj->tail = entry;
 211 }
 212
 213 /* remove a thread from an object wait queue */
 214 void remove_queue( struct object *obj, struct wait_queue_entry *entry )
 215 {
 216     if (entry->next) entry->next->prev = entry->prev;
 217     else obj->tail = entry->prev;
 218     if (entry->prev) entry->prev->next = entry->next;
 219     else obj->head = entry->next;
 220     release_object( obj );
 221 }
 222
 223 /* finish waiting */
 224 static void end_wait( struct thread *thread )
 225 {
 226     struct thread_wait *wait = thread->wait;
 227     struct wait_queue_entry *entry;
 228     int i;
 229
 230     assert( wait );
 231     for (i = 0, entry = wait->queues; i < wait->count; i++, entry++)
 232         entry->obj->ops->remove_queue( entry->obj, entry );
 233     if (wait->flags & SELECT_TIMEOUT) set_select_timeout( thread->client_fd, NULL );
 234     free( wait );
 235     thread->wait = NULL;
 236 }
 237
 238 /* build the thread wait structure */
 239 static int wait_on( struct thread *thread, int count,
 240                     int *handles, int flags, int timeout )
 241 {
 242     struct thread_wait *wait;
 243     struct wait_queue_entry *entry;
 244     struct object *obj;
 245     int i;
 246
 247     if ((count < 0) || (count > MAXIMUM_WAIT_OBJECTS))
 248     {
 249         SET_ERROR( ERROR_INVALID_PARAMETER );
 250         return 0;
 251     }
 252     if (!(wait = mem_alloc( sizeof(*wait) + (count-1) * sizeof(*entry) ))) return 0;
 253     thread->wait  = wait;
 254     wait->count   = count;
 255     wait->flags   = flags;
 256     if (flags & SELECT_TIMEOUT)
 257     {
 258         gettimeofday( &wait->timeout, 0 );
 259         if (timeout)
 260         {
 261             wait->timeout.tv_usec += (timeout % 1000) * 1000;
 262             if (wait->timeout.tv_usec >= 1000000)
 263             {
 264                 wait->timeout.tv_usec -= 1000000;
 265                 wait->timeout.tv_sec++;
 266             }
 267             wait->timeout.tv_sec += timeout / 1000;
 268         }
 269     }
 270
 271     for (i = 0, entry = wait->queues; i < count; i++, entry++)
 272     {
 273         if (!(obj = get_handle_obj( thread->process, handles[i],
 274                                     SYNCHRONIZE, NULL )))
 275         {
 276             wait->count = i - 1;
 277             end_wait( thread );
 278             return 0;
 279         }
 280         entry->thread = thread;
 281         obj->ops->add_queue( obj, entry );
 282         release_object( obj );
 283     }
 284     return 1;
 285 }
 286
 287 /* check if the thread waiting condition is satisfied */
 288 static int check_wait( struct thread *thread, int *signaled )
 289 {
 290     int i;
 291     struct thread_wait *wait = thread->wait;
 292     struct wait_queue_entry *entry = wait->queues;
 293     struct timeval now;
 294
 295     assert( wait );
 296     if (wait->flags & SELECT_ALL)
 297     {
 298         for (i = 0, entry = wait->queues; i < wait->count; i++, entry++)
 299             if (!entry->obj->ops->signaled( entry->obj, thread )) goto check_timeout;
 300         /* Wait satisfied: tell it to all objects */
 301         *signaled = 0;
 302         for (i = 0, entry = wait->queues; i < wait->count; i++, entry++)
 303             if (entry->obj->ops->satisfied( entry->obj, thread ))
 304                 *signaled = STATUS_ABANDONED_WAIT_0;
 305         return 1;
 306     }
 307     else
 308     {
 309         for (i = 0, entry = wait->queues; i < wait->count; i++, entry++)
 310         {
 311             if (!entry->obj->ops->signaled( entry->obj, thread )) continue;
 312             /* Wait satisfied: tell it to the object */
 313             *signaled = i;
 314             if (entry->obj->ops->satisfied( entry->obj, thread ))
 315                 *signaled += STATUS_ABANDONED_WAIT_0;
 316             return 1;
 317         }
 318     }
 319  check_timeout:
 320     if (!(wait->flags & SELECT_TIMEOUT)) return 0;
 321     gettimeofday( &now, NULL );
 322     if ((now.tv_sec > wait->timeout.tv_sec) ||
 323         ((now.tv_sec == wait->timeout.tv_sec) &&
 324          (now.tv_usec >= wait->timeout.tv_usec)))
 325     {
 326         *signaled = STATUS_TIMEOUT;
 327         return 1;
 328     }
 329     return 0;
 330 }
 331
 332 /* sleep on a list of objects */
 333 void sleep_on( struct thread *thread, int count, int *handles, int flags, int timeout )
 334 {
 335     struct select_reply reply;
 336
 337     assert( !thread->wait );
 338     reply.signaled = -1;
 339     if (!wait_on( thread, count, handles, flags, timeout )) goto done;
 340     if (!check_wait( thread, &reply.signaled ))
 341     {
 342         /* we need to wait */
 343         if (flags & SELECT_TIMEOUT)
 344             set_select_timeout( thread->client_fd, &thread->wait->timeout );
 345         return;
 346     }
 347     end_wait( thread );
 348  done:
 349     send_reply( thread, -1, 1, &reply, sizeof(reply) );
 350 }
 351
 352 /* attempt to wake up a thread */
 353 /* return 1 if OK, 0 if the wait condition is still not satisfied */
 354 static int wake_thread( struct thread *thread )
 355 {
 356     struct select_reply reply;
 357
 358     if (!check_wait( thread, &reply.signaled )) return 0;
 359     end_wait( thread );
 360     send_reply( thread, -1, 1, &reply, sizeof(reply) );
 361     return 1;
 362 }
 363
 364 /* timeout for the current thread */
 365 void thread_timeout(void)
 366 {
 367     struct select_reply reply;
 368
 369     assert( current->wait );
 370
 371     reply.signaled = STATUS_TIMEOUT;
 372     end_wait( current );
 373     send_reply( current, -1, 1, &reply, sizeof(reply) );
 374 }
 375
 376 /* attempt to wake threads sleeping on the object wait queue */
 377 void wake_up( struct object *obj, int max )
 378 {
 379     struct wait_queue_entry *entry = obj->head;
 380
 381     while (entry)
 382     {
 383         struct wait_queue_entry *next = entry->next;
 384         if (wake_thread( entry->thread ))
 385         {
 386             if (max && !--max) break;
 387         }
 388         entry = next;
 389     }
 390 }
 391
 392 /* kill a thread on the spot */
 393 void kill_thread( struct thread *thread, int exit_code )
 394 {
 395     if (thread->state == TERMINATED) return;  /* already killed */
 396     if (thread->unix_pid) kill( thread->unix_pid, SIGTERM );
 397     remove_client( thread->client_fd, exit_code ); /* this will call thread_killed */
 398 }
 399
 400 /* a thread has been killed */
 401 void thread_killed( struct thread *thread, int exit_code )
 402 {
 403     thread->state = TERMINATED;
 404     thread->exit_code = exit_code;
 405     if (thread->wait) end_wait( thread );
 406     abandon_mutexes( thread );
 407     remove_process_thread( thread->process, thread );
 408     wake_up( &thread->obj, 0 );
 409     release_object( thread );
 410 }