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