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