server/thread.c

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