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