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 #include <stdarg.h>
  17
  18
  19 #include "winbase.h"
  20 #include "winerror.h"
  21 #include "server.h"
  22 #include "server/thread.h"
  23 #include "server/process.h"
  24
  25
  26 /* thread queues */
  27
  28 struct wait_queue_entry
  29 {
  30     struct wait_queue_entry *next;
  31     struct wait_queue_entry *prev;
  32     struct object           *obj;
  33     struct thread           *thread;
  34 };
  35
  36 struct thread_wait
  37 {
  38     int                     count;      /* count of objects */
  39     int                     flags;
  40     struct timeval          timeout;
  41     struct wait_queue_entry queues[1];
  42 };
  43
  44 /* asynchronous procedure calls */
  45
  46 struct thread_apc
  47 {
  48     void                   *func;    /* function to call in client */
  49     void                   *param;   /* function param */
  50 };
  51 #define MAX_THREAD_APC  16  /* Max outstanding APCs for a thread */
  52
  53
  54 /* thread operations */
  55
  56 static void dump_thread( struct object *obj, int verbose );
  57 static int thread_signaled( struct object *obj, struct thread *thread );
  58 static void destroy_thread( struct object *obj );
  59
  60 static const struct object_ops thread_ops =
  61 {
  62     dump_thread,
  63     add_queue,
  64     remove_queue,
  65     thread_signaled,
  66     no_satisfied,
  67     no_read_fd,
  68     no_write_fd,
  69     no_flush,
  70     no_get_file_info,
  71     destroy_thread
  72 };
  73
  74 static struct thread *first_thread;
  75
  76
  77 /* create a new thread */
  78 struct thread *create_thread( int fd, void *pid, int *thread_handle,
  79                               int *process_handle )
  80 {
  81     struct thread *thread;
  82     struct process *process;
  83
  84     if (!(thread = mem_alloc( sizeof(*thread) ))) return NULL;
  85
  86     if (pid) process = get_process_from_id( pid );
  87     else process = create_process();
  88     if (!process)
  89     {
  90         free( thread );
  91         return NULL;
  92     }
  93
  94     init_object( &thread->obj, &thread_ops, NULL );
  95     thread->client_fd = fd;
  96     thread->process   = process;
  97     thread->unix_pid  = 0;  /* not known yet */
  98     thread->name      = NULL;
  99     thread->mutex     = NULL;
 100     thread->wait      = NULL;
 101     thread->apc       = NULL;
 102     thread->apc_count = 0;
 103     thread->error     = 0;
 104     thread->state     = STARTING;
 105     thread->exit_code = 0x103;  /* STILL_ACTIVE */
 106     thread->next      = first_thread;
 107     thread->prev      = NULL;
 108     thread->priority  = THREAD_PRIORITY_NORMAL;
 109     thread->affinity  = 1;
 110     thread->suspend   = 0;
 111
 112     if (first_thread) first_thread->prev = thread;
 113     first_thread = thread;
 114     add_process_thread( process, thread );
 115
 116     *thread_handle = *process_handle = -1;
 117     if (current)
 118     {
 119         if ((*thread_handle = alloc_handle( current->process, thread,
 120                                             THREAD_ALL_ACCESS, 0 )) == -1)
 121             goto error;
 122     }
 123     if (current && !pid)
 124     {
 125         if ((*process_handle = alloc_handle( current->process, process,
 126                                              PROCESS_ALL_ACCESS, 0 )) == -1)
 127             goto error;
 128     }
 129
 130     if (add_client( fd, thread ) == -1) goto error;
 131
 132     return thread;
 133
 134  error:
 135     if (current)
 136     {
 137         close_handle( current->process, *thread_handle );
 138         close_handle( current->process, *process_handle );
 139     }
 140     remove_process_thread( process, thread );
 141     release_object( thread );
 142     return NULL;
 143 }
 144
 145 /* destroy a thread when its refcount is 0 */
 146 static void destroy_thread( struct object *obj )
 147 {
 148     struct thread *thread = (struct thread *)obj;
 149     assert( obj->ops == &thread_ops );
 150
 151     release_object( thread->process );
 152     if (thread->next) thread->next->prev = thread->prev;
 153     if (thread->prev) thread->prev->next = thread->next;
 154     else first_thread = thread->next;
 155     if (thread->name) free( thread->name );
 156     if (thread->apc) free( thread->apc );
 157     if (debug_level) memset( thread, 0xaa, sizeof(thread) );  /* catch errors */
 158     free( thread );
 159 }
 160
 161 /* dump a thread on stdout for debugging purposes */
 162 static void dump_thread( struct object *obj, int verbose )
 163 {
 164     struct thread *thread = (struct thread *)obj;
 165     assert( obj->ops == &thread_ops );
 166
 167     fprintf( stderr, "Thread pid=%d fd=%d name='%s'\n",
 168              thread->unix_pid, thread->client_fd, thread->name );
 169 }
 170
 171 static int thread_signaled( struct object *obj, struct thread *thread )
 172 {
 173     struct thread *mythread = (struct thread *)obj;
 174     return (mythread->state == TERMINATED);
 175 }
 176
 177 /* get a thread pointer from a thread id (and increment the refcount) */
 178 struct thread *get_thread_from_id( void *id )
 179 {
 180     struct thread *t = first_thread;
 181     while (t && (t != id)) t = t->next;
 182     if (t) grab_object( t );
 183     return t;
 184 }
 185
 186 /* get a thread from a handle (and increment the refcount) */
 187 struct thread *get_thread_from_handle( int handle, unsigned int access )
 188 {
 189     return (struct thread *)get_handle_obj( current->process, handle,
 190                                             access, &thread_ops );
 191 }
 192
 193 /* get all information about a thread */
 194 void get_thread_info( struct thread *thread,
 195                       struct get_thread_info_reply *reply )
 196 {
 197     reply->pid       = thread;
 198     reply->exit_code = thread->exit_code;
 199     reply->priority  = thread->priority;
 200 }
 201
 202
 203 /* set all information about a thread */
 204 void set_thread_info( struct thread *thread,
 205                       struct set_thread_info_request *req )
 206 {
 207     if (req->mask & SET_THREAD_INFO_PRIORITY)
 208         thread->priority = req->priority;
 209     if (req->mask & SET_THREAD_INFO_AFFINITY)
 210     {
 211         if (req->affinity != 1) SET_ERROR( ERROR_INVALID_PARAMETER );
 212         else thread->affinity = req->affinity;
 213     }
 214 }
 215
 216 /* suspend a thread */
 217 int suspend_thread( struct thread *thread )
 218 {
 219     int old_count = thread->suspend;
 220     if (thread->suspend < MAXIMUM_SUSPEND_COUNT)
 221     {
 222         if (!thread->suspend++)
 223         {
 224             if (thread->unix_pid) kill( thread->unix_pid, SIGSTOP );
 225         }
 226     }
 227     return old_count;
 228 }
 229
 230 /* resume a thread */
 231 int resume_thread( struct thread *thread )
 232 {
 233     int old_count = thread->suspend;
 234     if (thread->suspend > 0)
 235     {
 236         if (!--thread->suspend)
 237         {
 238             if (thread->unix_pid) kill( thread->unix_pid, SIGCONT );
 239         }
 240     }
 241     return old_count;
 242 }
 243
 244 /* send a reply to a thread */
 245 int send_reply( struct thread *thread, int pass_fd, int n,
 246                 ... /* arg_1, len_1, ..., arg_n, len_n */ )
 247 {
 248     struct iovec vec[16];
 249     va_list args;
 250     int i;
 251
 252     assert( n < 16 );
 253     va_start( args, n );
 254     for (i = 0; i < n; i++)
 255     {
 256         vec[i].iov_base = va_arg( args, void * );
 257         vec[i].iov_len  = va_arg( args, int );
 258     }
 259     va_end( args );
 260     return send_reply_v( thread->client_fd, thread->error, pass_fd, vec, n );
 261 }
 262
 263 /* add a thread to an object wait queue; return 1 if OK, 0 on error */
 264 int add_queue( struct object *obj, struct wait_queue_entry *entry )
 265 {
 266     grab_object( obj );
 267     entry->obj    = obj;
 268     entry->prev   = obj->tail;
 269     entry->next   = NULL;
 270     if (obj->tail) obj->tail->next = entry;
 271     else obj->head = entry;
 272     obj->tail = entry;
 273     return 1;
 274 }
 275
 276 /* remove a thread from an object wait queue */
 277 void remove_queue( struct object *obj, struct wait_queue_entry *entry )
 278 {
 279     if (entry->next) entry->next->prev = entry->prev;
 280     else obj->tail = entry->prev;
 281     if (entry->prev) entry->prev->next = entry->next;
 282     else obj->head = entry->next;
 283     release_object( obj );
 284 }
 285
 286 /* finish waiting */
 287 static void end_wait( struct thread *thread )
 288 {
 289     struct thread_wait *wait = thread->wait;
 290     struct wait_queue_entry *entry;
 291     int i;
 292
 293     assert( wait );
 294     for (i = 0, entry = wait->queues; i < wait->count; i++, entry++)
 295         entry->obj->ops->remove_queue( entry->obj, entry );
 296     if (wait->flags & SELECT_TIMEOUT) set_select_timeout( thread->client_fd, NULL );
 297     free( wait );
 298     thread->wait = NULL;
 299 }
 300
 301 /* build the thread wait structure */
 302 static int wait_on( struct thread *thread, int count,
 303                     int *handles, int flags, int timeout )
 304 {
 305     struct thread_wait *wait;
 306     struct wait_queue_entry *entry;
 307     struct object *obj;
 308     int i;
 309
 310     if ((count < 0) || (count > MAXIMUM_WAIT_OBJECTS))
 311     {
 312         SET_ERROR( ERROR_INVALID_PARAMETER );
 313         return 0;
 314     }
 315     if (!(wait = mem_alloc( sizeof(*wait) + (count-1) * sizeof(*entry) ))) return 0;
 316     thread->wait  = wait;
 317     wait->count   = count;
 318     wait->flags   = flags;
 319     if (flags & SELECT_TIMEOUT)
 320     {
 321         gettimeofday( &wait->timeout, 0 );
 322         if (timeout)
 323         {
 324             wait->timeout.tv_usec += (timeout % 1000) * 1000;
 325             if (wait->timeout.tv_usec >= 1000000)
 326             {
 327                 wait->timeout.tv_usec -= 1000000;
 328                 wait->timeout.tv_sec++;
 329             }
 330             wait->timeout.tv_sec += timeout / 1000;
 331         }
 332     }
 333
 334     for (i = 0, entry = wait->queues; i < count; i++, entry++)
 335     {
 336         if (!(obj = get_handle_obj( thread->process, handles[i],
 337                                     SYNCHRONIZE, NULL )))
 338         {
 339             wait->count = i - 1;
 340             end_wait( thread );
 341             return 0;
 342         }
 343         entry->thread = thread;
 344         if (!obj->ops->add_queue( obj, entry ))
 345         {
 346             wait->count = i - 1;
 347             end_wait( thread );
 348             return 0;
 349         }
 350         release_object( obj );
 351     }
 352     return 1;
 353 }
 354
 355 /* check if the thread waiting condition is satisfied */
 356 static int check_wait( struct thread *thread, int *signaled )
 357 {
 358     int i;
 359     struct thread_wait *wait = thread->wait;
 360     struct wait_queue_entry *entry = wait->queues;
 361
 362     assert( wait );
 363     if (wait->flags & SELECT_ALL)
 364     {
 365         for (i = 0, entry = wait->queues; i < wait->count; i++, entry++)
 366             if (!entry->obj->ops->signaled( entry->obj, thread )) goto other_checks;
 367         /* Wait satisfied: tell it to all objects */
 368         *signaled = 0;
 369         for (i = 0, entry = wait->queues; i < wait->count; i++, entry++)
 370             if (entry->obj->ops->satisfied( entry->obj, thread ))
 371                 *signaled = STATUS_ABANDONED_WAIT_0;
 372         return 1;
 373     }
 374     else
 375     {
 376         for (i = 0, entry = wait->queues; i < wait->count; i++, entry++)
 377         {
 378             if (!entry->obj->ops->signaled( entry->obj, thread )) continue;
 379             /* Wait satisfied: tell it to the object */
 380             *signaled = i;
 381             if (entry->obj->ops->satisfied( entry->obj, thread ))
 382                 *signaled += STATUS_ABANDONED_WAIT_0;
 383             return 1;
 384         }
 385     }
 386
 387  other_checks:
 388     if ((wait->flags & SELECT_ALERTABLE) && thread->apc)
 389     {
 390         *signaled = STATUS_USER_APC;
 391         return 1;
 392     }
 393     if (wait->flags & SELECT_TIMEOUT)
 394     {
 395         struct timeval now;
 396         gettimeofday( &now, NULL );
 397         if ((now.tv_sec > wait->timeout.tv_sec) ||
 398             ((now.tv_sec == wait->timeout.tv_sec) &&
 399              (now.tv_usec >= wait->timeout.tv_usec)))
 400         {
 401             *signaled = STATUS_TIMEOUT;
 402             return 1;
 403         }
 404     }
 405     return 0;
 406 }
 407
 408 /* send the select reply to wake up the client */
 409 static void send_select_reply( struct thread *thread, int signaled )
 410 {
 411     struct select_reply reply;
 412     reply.signaled = signaled;
 413     if ((signaled == STATUS_USER_APC) && thread->apc)
 414     {
 415         struct thread_apc *apc = thread->apc;
 416         int len = thread->apc_count * sizeof(*apc);
 417         thread->apc = NULL;
 418         thread->apc_count = 0;
 419         send_reply( thread, -1, 2, &reply, sizeof(reply),
 420                     apc, len );
 421         free( apc );
 422     }
 423     else send_reply( thread, -1, 1, &reply, sizeof(reply) );
 424 }
 425
 426 /* attempt to wake up a thread */
 427 /* return 1 if OK, 0 if the wait condition is still not satisfied */
 428 static int wake_thread( struct thread *thread )
 429 {
 430     int signaled;
 431
 432     if (!check_wait( thread, &signaled )) return 0;
 433     end_wait( thread );
 434     send_select_reply( thread, signaled );
 435     return 1;
 436 }
 437
 438 /* sleep on a list of objects */
 439 void sleep_on( struct thread *thread, int count, int *handles, int flags, int timeout )
 440 {
 441     assert( !thread->wait );
 442     if (!wait_on( thread, count, handles, flags, timeout ))
 443     {
 444         /* return an error */
 445         send_select_reply( thread, -1 );
 446         return;
 447     }
 448     if (!wake_thread( thread ))
 449     {
 450         /* we need to wait */
 451         if (flags & SELECT_TIMEOUT)
 452             set_select_timeout( thread->client_fd, &thread->wait->timeout );
 453     }
 454 }
 455
 456 /* timeout for the current thread */
 457 void thread_timeout(void)
 458 {
 459     assert( current->wait );
 460     end_wait( current );
 461     send_select_reply( current, STATUS_TIMEOUT );
 462 }
 463
 464 /* attempt to wake threads sleeping on the object wait queue */
 465 void wake_up( struct object *obj, int max )
 466 {
 467     struct wait_queue_entry *entry = obj->head;
 468
 469     while (entry)
 470     {
 471         struct wait_queue_entry *next = entry->next;
 472         if (wake_thread( entry->thread ))
 473         {
 474             if (max && !--max) break;
 475         }
 476         entry = next;
 477     }
 478 }
 479
 480 /* queue an async procedure call */
 481 int thread_queue_apc( struct thread *thread, void *func, void *param )
 482 {
 483     struct thread_apc *apc;
 484     if (!func)
 485     {
 486         SET_ERROR( ERROR_INVALID_PARAMETER );
 487         return 0;
 488     }
 489     if (!thread->apc)
 490     {
 491         if (!(thread->apc = mem_alloc( MAX_THREAD_APC * sizeof(*apc) )))
 492             return 0;
 493         thread->apc_count = 0;
 494     }
 495     else if (thread->apc_count >= MAX_THREAD_APC) return 0;
 496     thread->apc[thread->apc_count].func  = func;
 497     thread->apc[thread->apc_count].param = param;
 498     thread->apc_count++;
 499     wake_thread( thread );
 500     return 1;
 501 }
 502
 503 /* kill a thread on the spot */
 504 void kill_thread( struct thread *thread, int exit_code )
 505 {
 506     if (thread->state == TERMINATED) return;  /* already killed */
 507     if (thread->unix_pid) kill( thread->unix_pid, SIGTERM );
 508     remove_client( thread->client_fd, exit_code ); /* this will call thread_killed */
 509 }
 510
 511 /* a thread has been killed */
 512 void thread_killed( struct thread *thread, int exit_code )
 513 {
 514     thread->state = TERMINATED;
 515     thread->exit_code = exit_code;
 516     if (thread->wait) end_wait( thread );
 517     abandon_mutexes( thread );
 518     remove_process_thread( thread->process, thread );
 519     wake_up( &thread->obj, 0 );
 520     release_object( thread );
 521 }