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
  22 #include "handle.h"
  23 #include "server.h"
  24 #include "process.h"
  25 #include "thread.h"
  26
  27
  28 /* thread queues */
  29
  30 struct wait_queue_entry
  31 {
  32     struct wait_queue_entry *next;
  33     struct wait_queue_entry *prev;
  34     struct object           *obj;
  35     struct thread           *thread;
  36 };
  37
  38 struct thread_wait
  39 {
  40     int                     count;      /* count of objects */
  41     int                     flags;
  42     struct timeval          timeout;
  43     struct timeout_user    *user;
  44     struct wait_queue_entry queues[1];
  45 };
  46
  47 /* asynchronous procedure calls */
  48
  49 struct thread_apc
  50 {
  51     void                   *func;    /* function to call in client */
  52     void                   *param;   /* function param */
  53 };
  54 #define MAX_THREAD_APC  16  /* Max outstanding APCs for a thread */
  55
  56
  57 /* thread operations */
  58
  59 static void dump_thread( struct object *obj, int verbose );
  60 static int thread_signaled( struct object *obj, struct thread *thread );
  61 static void destroy_thread( struct object *obj );
  62
  63 static const struct object_ops thread_ops =
  64 {
  65     dump_thread,
  66     add_queue,
  67     remove_queue,
  68     thread_signaled,
  69     no_satisfied,
  70     no_read_fd,
  71     no_write_fd,
  72     no_flush,
  73     no_get_file_info,
  74     destroy_thread
  75 };
  76
  77 static struct thread *first_thread;
  78
  79 /* initialization of a thread structure */
  80 static void init_thread( struct thread *thread, int fd )
  81 {
  82     init_object( &thread->obj, &thread_ops, NULL );
  83     thread->client      = NULL;
  84     thread->unix_pid    = 0;  /* not known yet */
  85     thread->teb         = NULL;
  86     thread->mutex       = NULL;
  87     thread->debug_ctx   = NULL;
  88     thread->debug_first = NULL;
  89     thread->debug_event = NULL;
  90     thread->wait        = NULL;
  91     thread->apc         = NULL;
  92     thread->apc_count   = 0;
  93     thread->error       = 0;
  94     thread->state       = STARTING;
  95     thread->exit_code   = 0x103;  /* STILL_ACTIVE */
  96     thread->next        = NULL;
  97     thread->prev        = NULL;
  98     thread->priority    = THREAD_PRIORITY_NORMAL;
  99     thread->affinity    = 1;
 100     thread->suspend     = 0;
 101 }
 102
 103 /* create the initial thread and start the main server loop */
 104 void create_initial_thread( int fd )
 105 {
 106     first_thread = mem_alloc( sizeof(*first_thread) );
 107     assert( first_thread );
 108
 109     current = first_thread;
 110     init_thread( first_thread, fd );
 111     first_thread->process = create_initial_process();
 112     add_process_thread( first_thread->process, first_thread );
 113     first_thread->client = add_client( fd, first_thread );
 114     select_loop();
 115 }
 116
 117 /* create a new thread */
 118 static struct thread *create_thread( int fd, void *pid, int suspend, int inherit, int *handle )
 119 {
 120     struct thread *thread;
 121     struct process *process;
 122
 123     if (!(thread = mem_alloc( sizeof(*thread) ))) return NULL;
 124
 125     if (!(process = get_process_from_id( pid )))
 126     {
 127         free( thread );
 128         return NULL;
 129     }
 130     init_thread( thread, fd );
 131     thread->process = process;
 132
 133     if (suspend) thread->suspend++;
 134
 135     if ((thread->next = first_thread) != NULL) thread->next->prev = thread;
 136     first_thread = thread;
 137     add_process_thread( process, thread );
 138
 139     if ((*handle = alloc_handle( current->process, thread,
 140                                  THREAD_ALL_ACCESS, inherit )) == -1) goto error;
 141     if (!(thread->client = add_client( fd, thread )))
 142     {
 143         SET_ERROR( ERROR_TOO_MANY_OPEN_FILES );
 144         goto error;
 145     }
 146     return thread;
 147
 148  error:
 149     close_handle( current->process, *handle );
 150     remove_process_thread( process, thread );
 151     release_object( thread );
 152     return NULL;
 153 }
 154
 155 /* destroy a thread when its refcount is 0 */
 156 static void destroy_thread( struct object *obj )
 157 {
 158     struct thread *thread = (struct thread *)obj;
 159     assert( obj->ops == &thread_ops );
 160
 161     assert( !thread->debug_ctx );  /* cannot still be debugging something */
 162     release_object( thread->process );
 163     if (thread->next) thread->next->prev = thread->prev;
 164     if (thread->prev) thread->prev->next = thread->next;
 165     else first_thread = thread->next;
 166     if (thread->apc) free( thread->apc );
 167     if (debug_level) memset( thread, 0xaa, sizeof(thread) );  /* catch errors */
 168     free( thread );
 169 }
 170
 171 /* dump a thread on stdout for debugging purposes */
 172 static void dump_thread( struct object *obj, int verbose )
 173 {
 174     struct thread *thread = (struct thread *)obj;
 175     assert( obj->ops == &thread_ops );
 176
 177     fprintf( stderr, "Thread pid=%d teb=%p state=%d\n",
 178              thread->unix_pid, thread->teb, thread->state );
 179 }
 180
 181 static int thread_signaled( struct object *obj, struct thread *thread )
 182 {
 183     struct thread *mythread = (struct thread *)obj;
 184     return (mythread->state == TERMINATED);
 185 }
 186
 187 /* get a thread pointer from a thread id (and increment the refcount) */
 188 struct thread *get_thread_from_id( void *id )
 189 {
 190     struct thread *t = first_thread;
 191     while (t && (t != id)) t = t->next;
 192     if (t) grab_object( t );
 193     return t;
 194 }
 195
 196 /* get a thread from a handle (and increment the refcount) */
 197 struct thread *get_thread_from_handle( int handle, unsigned int access )
 198 {
 199     return (struct thread *)get_handle_obj( current->process, handle,
 200                                             access, &thread_ops );
 201 }
 202
 203 /* set all information about a thread */
 204 static 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 static int suspend_thread( struct thread *thread )
 218 {
 219     int old_count = thread->suspend;
 220     if (thread->suspend < MAXIMUM_SUSPEND_COUNT)
 221     {
 222         if (!(thread->process->suspend + 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 static int resume_thread( struct thread *thread )
 232 {
 233     int old_count = thread->suspend;
 234     if (thread->suspend > 0)
 235     {
 236         if (!(--thread->suspend + thread->process->suspend))
 237         {
 238             if (thread->unix_pid) kill( thread->unix_pid, SIGCONT );
 239         }
 240     }
 241     return old_count;
 242 }
 243
 244 /* suspend all threads but the current */
 245 void suspend_all_threads( void )
 246 {
 247     struct thread *thread;
 248     for ( thread = first_thread; thread; thread = thread->next )
 249         if ( thread != current )
 250             suspend_thread( thread );
 251 }
 252
 253 /* resume all threads but the current */
 254 void resume_all_threads( void )
 255 {
 256     struct thread *thread;
 257     for ( thread = first_thread; thread; thread = thread->next )
 258         if ( thread != current )
 259             resume_thread( thread );
 260 }
 261
 262 /* send a reply to a thread */
 263 int send_reply( struct thread *thread, int pass_fd, int n,
 264                 ... /* arg_1, len_1, ..., arg_n, len_n */ )
 265 {
 266     struct iovec vec[16];
 267     va_list args;
 268     int i;
 269
 270     assert( n < 16 );
 271     va_start( args, n );
 272     for (i = 0; i < n; i++)
 273     {
 274         vec[i].iov_base = va_arg( args, void * );
 275         vec[i].iov_len  = va_arg( args, int );
 276     }
 277     va_end( args );
 278     return send_reply_v( thread->client, thread->error, pass_fd, vec, n );
 279 }
 280
 281 /* add a thread to an object wait queue; return 1 if OK, 0 on error */
 282 int add_queue( struct object *obj, struct wait_queue_entry *entry )
 283 {
 284     grab_object( obj );
 285     entry->obj    = obj;
 286     entry->prev   = obj->tail;
 287     entry->next   = NULL;
 288     if (obj->tail) obj->tail->next = entry;
 289     else obj->head = entry;
 290     obj->tail = entry;
 291     return 1;
 292 }
 293
 294 /* remove a thread from an object wait queue */
 295 void remove_queue( struct object *obj, struct wait_queue_entry *entry )
 296 {
 297     if (entry->next) entry->next->prev = entry->prev;
 298     else obj->tail = entry->prev;
 299     if (entry->prev) entry->prev->next = entry->next;
 300     else obj->head = entry->next;
 301     release_object( obj );
 302 }
 303
 304 /* finish waiting */
 305 static void end_wait( struct thread *thread )
 306 {
 307     struct thread_wait *wait = thread->wait;
 308     struct wait_queue_entry *entry;
 309     int i;
 310
 311     assert( wait );
 312     for (i = 0, entry = wait->queues; i < wait->count; i++, entry++)
 313         entry->obj->ops->remove_queue( entry->obj, entry );
 314     if (wait->user) remove_timeout_user( wait->user );
 315     free( wait );
 316     thread->wait = NULL;
 317 }
 318
 319 /* build the thread wait structure */
 320 static int wait_on( struct thread *thread, int count,
 321                     int *handles, int flags, int timeout )
 322 {
 323     struct thread_wait *wait;
 324     struct wait_queue_entry *entry;
 325     struct object *obj;
 326     int i;
 327
 328     if ((count < 0) || (count > MAXIMUM_WAIT_OBJECTS))
 329     {
 330         SET_ERROR( ERROR_INVALID_PARAMETER );
 331         return 0;
 332     }
 333     if (!(wait = mem_alloc( sizeof(*wait) + (count-1) * sizeof(*entry) ))) return 0;
 334     thread->wait  = wait;
 335     wait->count   = count;
 336     wait->flags   = flags;
 337     wait->user    = NULL;
 338     if (flags & SELECT_TIMEOUT) make_timeout( &wait->timeout, timeout );
 339
 340     for (i = 0, entry = wait->queues; i < count; i++, entry++)
 341     {
 342         if (!(obj = get_handle_obj( thread->process, handles[i],
 343                                     SYNCHRONIZE, NULL )))
 344         {
 345             wait->count = i - 1;
 346             end_wait( thread );
 347             return 0;
 348         }
 349         entry->thread = thread;
 350         if (!obj->ops->add_queue( obj, entry ))
 351         {
 352             wait->count = i - 1;
 353             end_wait( thread );
 354             return 0;
 355         }
 356         release_object( obj );
 357     }
 358     return 1;
 359 }
 360
 361 /* check if the thread waiting condition is satisfied */
 362 static int check_wait( struct thread *thread, int *signaled )
 363 {
 364     int i;
 365     struct thread_wait *wait = thread->wait;
 366     struct wait_queue_entry *entry = wait->queues;
 367
 368     assert( wait );
 369     if (wait->flags & SELECT_ALL)
 370     {
 371         int not_ok = 0;
 372         /* Note: we must check them all anyway, as some objects may
 373          * want to do something when signaled, even if others are not */
 374         for (i = 0, entry = wait->queues; i < wait->count; i++, entry++)
 375             not_ok |= !entry->obj->ops->signaled( entry->obj, thread );
 376         if (not_ok) goto other_checks;
 377         /* Wait satisfied: tell it to all objects */
 378         *signaled = 0;
 379         for (i = 0, entry = wait->queues; i < wait->count; i++, entry++)
 380             if (entry->obj->ops->satisfied( entry->obj, thread ))
 381                 *signaled = STATUS_ABANDONED_WAIT_0;
 382         return 1;
 383     }
 384     else
 385     {
 386         for (i = 0, entry = wait->queues; i < wait->count; i++, entry++)
 387         {
 388             if (!entry->obj->ops->signaled( entry->obj, thread )) continue;
 389             /* Wait satisfied: tell it to the object */
 390             *signaled = i;
 391             if (entry->obj->ops->satisfied( entry->obj, thread ))
 392                 *signaled += STATUS_ABANDONED_WAIT_0;
 393             return 1;
 394         }
 395     }
 396
 397  other_checks:
 398     if ((wait->flags & SELECT_ALERTABLE) && thread->apc)
 399     {
 400         *signaled = STATUS_USER_APC;
 401         return 1;
 402     }
 403     if (wait->flags & SELECT_TIMEOUT)
 404     {
 405         struct timeval now;
 406         gettimeofday( &now, NULL );
 407         if ((now.tv_sec > wait->timeout.tv_sec) ||
 408             ((now.tv_sec == wait->timeout.tv_sec) &&
 409              (now.tv_usec >= wait->timeout.tv_usec)))
 410         {
 411             *signaled = STATUS_TIMEOUT;
 412             return 1;
 413         }
 414     }
 415     return 0;
 416 }
 417
 418 /* send the select reply to wake up the client */
 419 static void send_select_reply( struct thread *thread, int signaled )
 420 {
 421     struct select_reply reply;
 422     reply.signaled = signaled;
 423     if ((signaled == STATUS_USER_APC) && thread->apc)
 424     {
 425         struct thread_apc *apc = thread->apc;
 426         int len = thread->apc_count * sizeof(*apc);
 427         thread->apc = NULL;
 428         thread->apc_count = 0;
 429         send_reply( thread, -1, 2, &reply, sizeof(reply),
 430                     apc, len );
 431         free( apc );
 432     }
 433     else send_reply( thread, -1, 1, &reply, sizeof(reply) );
 434 }
 435
 436 /* attempt to wake up a thread */
 437 /* return 1 if OK, 0 if the wait condition is still not satisfied */
 438 static int wake_thread( struct thread *thread )
 439 {
 440     int signaled;
 441
 442     if (!check_wait( thread, &signaled )) return 0;
 443     end_wait( thread );
 444     send_select_reply( thread, signaled );
 445     return 1;
 446 }
 447
 448 /* sleep on a list of objects */
 449 static void sleep_on( struct thread *thread, int count, int *handles, int flags, int timeout )
 450 {
 451     assert( !thread->wait );
 452     if (!wait_on( thread, count, handles, flags, timeout ))
 453     {
 454         /* return an error */
 455         send_select_reply( thread, -1 );
 456         return;
 457     }
 458     if (wake_thread( thread )) return;
 459     /* now we need to wait */
 460     if (flags & SELECT_TIMEOUT)
 461     {
 462         if (!(thread->wait->user = add_timeout_user( &thread->wait->timeout,
 463                                                      call_timeout_handler, thread )))
 464         {
 465             send_select_reply( thread, -1 );
 466         }
 467     }
 468 }
 469
 470 /* timeout for the current thread */
 471 void thread_timeout(void)
 472 {
 473     assert( current->wait );
 474     current->wait->user = NULL;
 475     end_wait( current );
 476     send_select_reply( current, STATUS_TIMEOUT );
 477 }
 478
 479 /* attempt to wake threads sleeping on the object wait queue */
 480 void wake_up( struct object *obj, int max )
 481 {
 482     struct wait_queue_entry *entry = obj->head;
 483
 484     while (entry)
 485     {
 486         struct wait_queue_entry *next = entry->next;
 487         if (wake_thread( entry->thread ))
 488         {
 489             if (max && !--max) break;
 490         }
 491         entry = next;
 492     }
 493 }
 494
 495 /* queue an async procedure call */
 496 static int thread_queue_apc( struct thread *thread, void *func, void *param )
 497 {
 498     struct thread_apc *apc;
 499     if (!thread->apc)
 500     {
 501         if (!(thread->apc = mem_alloc( MAX_THREAD_APC * sizeof(*apc) )))
 502             return 0;
 503         thread->apc_count = 0;
 504     }
 505     else if (thread->apc_count >= MAX_THREAD_APC) return 0;
 506     thread->apc[thread->apc_count].func  = func;
 507     thread->apc[thread->apc_count].param = param;
 508     thread->apc_count++;
 509     if (thread->wait) wake_thread( thread );
 510     return 1;
 511 }
 512
 513 /* kill a thread on the spot */
 514 void kill_thread( struct thread *thread, int exit_code )
 515 {
 516     if (thread->state == TERMINATED) return;  /* already killed */
 517     if (thread->unix_pid) kill( thread->unix_pid, SIGTERM );
 518     remove_client( thread->client, exit_code ); /* this will call thread_killed */
 519 }
 520
 521 /* a thread has been killed */
 522 void thread_killed( struct thread *thread, int exit_code )
 523 {
 524     thread->state = TERMINATED;
 525     thread->exit_code = exit_code;
 526     if (thread->wait) end_wait( thread );
 527     debug_exit_thread( thread, exit_code );
 528     abandon_mutexes( thread );
 529     remove_process_thread( thread->process, thread );
 530     wake_up( &thread->obj, 0 );
 531     release_object( thread );
 532 }
 533
 534 /* create a new thread */
 535 DECL_HANDLER(new_thread)
 536 {
 537     struct new_thread_reply reply;
 538     int new_fd;
 539
 540     if ((new_fd = dup(fd)) != -1)
 541     {
 542         reply.tid = create_thread( new_fd, req->pid, req->suspend,
 543                                    req->inherit, &reply.handle );
 544         if (!reply.tid) close( new_fd );
 545     }
 546     else
 547         SET_ERROR( ERROR_TOO_MANY_OPEN_FILES );
 548
 549     send_reply( current, -1, 1, &reply, sizeof(reply) );
 550 }
 551
 552 /* initialize a new thread */
 553 DECL_HANDLER(init_thread)
 554 {
 555     struct init_thread_reply reply;
 556
 557     if (current->state != STARTING)
 558     {
 559         fatal_protocol_error( "init_thread: already running\n" );
 560         return;
 561     }
 562     current->state    = RUNNING;
 563     current->unix_pid = req->unix_pid;
 564     current->teb      = req->teb;
 565     if (current->suspend + current->process->suspend > 0)
 566         kill( current->unix_pid, SIGSTOP );
 567     reply.pid = current->process;
 568     reply.tid = current;
 569     send_reply( current, -1, 1, &reply, sizeof(reply) );
 570 }
 571
 572 /* terminate a thread */
 573 DECL_HANDLER(terminate_thread)
 574 {
 575     struct thread *thread;
 576
 577     if ((thread = get_thread_from_handle( req->handle, THREAD_TERMINATE )))
 578     {
 579         kill_thread( thread, req->exit_code );
 580         release_object( thread );
 581     }
 582     if (current) send_reply( current, -1, 0 );
 583 }
 584
 585 /* fetch information about a thread */
 586 DECL_HANDLER(get_thread_info)
 587 {
 588     struct thread *thread;
 589     struct get_thread_info_reply reply = { 0, 0, 0 };
 590
 591     if ((thread = get_thread_from_handle( req->handle, THREAD_QUERY_INFORMATION )))
 592     {
 593         reply.tid       = thread;
 594         reply.exit_code = thread->exit_code;
 595         reply.priority  = thread->priority;
 596         release_object( thread );
 597     }
 598     send_reply( current, -1, 1, &reply, sizeof(reply) );
 599 }
 600
 601 /* set information about a thread */
 602 DECL_HANDLER(set_thread_info)
 603 {
 604     struct thread *thread;
 605
 606     if ((thread = get_thread_from_handle( req->handle, THREAD_SET_INFORMATION )))
 607     {
 608         set_thread_info( thread, req );
 609         release_object( thread );
 610     }
 611     send_reply( current, -1, 0 );
 612 }
 613
 614 /* suspend a thread */
 615 DECL_HANDLER(suspend_thread)
 616 {
 617     struct thread *thread;
 618     struct suspend_thread_reply reply = { -1 };
 619     if ((thread = get_thread_from_handle( req->handle, THREAD_SUSPEND_RESUME )))
 620     {
 621         reply.count = suspend_thread( thread );
 622         release_object( thread );
 623     }
 624     send_reply( current, -1, 1, &reply, sizeof(reply) );
 625
 626 }
 627
 628 /* resume a thread */
 629 DECL_HANDLER(resume_thread)
 630 {
 631     struct thread *thread;
 632     struct resume_thread_reply reply = { -1 };
 633     if ((thread = get_thread_from_handle( req->handle, THREAD_SUSPEND_RESUME )))
 634     {
 635         reply.count = resume_thread( thread );
 636         release_object( thread );
 637     }
 638     send_reply( current, -1, 1, &reply, sizeof(reply) );
 639
 640 }
 641
 642 /* select on a handle list */
 643 DECL_HANDLER(select)
 644 {
 645     if (len != req->count * sizeof(int))
 646         fatal_protocol_error( "select: bad length %d for %d handles\n",
 647                               len, req->count );
 648     sleep_on( current, req->count, (int *)data, req->flags, req->timeout );
 649 }
 650
 651 /* queue an APC for a thread */
 652 DECL_HANDLER(queue_apc)
 653 {
 654     struct thread *thread;
 655     if ((thread = get_thread_from_handle( req->handle, THREAD_SET_CONTEXT )))
 656     {
 657         thread_queue_apc( thread, req->func, req->param );
 658         release_object( thread );
 659     }
 660     send_reply( current, -1, 0 );
 661 }