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 wait_queue_entry queues[1];
  44 };
  45
  46 /* asynchronous procedure calls */
  47
  48 struct thread_apc
  49 {
  50     void                   *func;    /* function to call in client */
  51     void                   *param;   /* function param */
  52 };
  53 #define MAX_THREAD_APC  16  /* Max outstanding APCs for a thread */
  54
  55
  56 /* thread operations */
  57
  58 static void dump_thread( struct object *obj, int verbose );
  59 static int thread_signaled( struct object *obj, struct thread *thread );
  60 static void destroy_thread( struct object *obj );
  61
  62 static const struct object_ops thread_ops =
  63 {
  64     dump_thread,
  65     add_queue,
  66     remove_queue,
  67     thread_signaled,
  68     no_satisfied,
  69     no_read_fd,
  70     no_write_fd,
  71     no_flush,
  72     no_get_file_info,
  73     destroy_thread
  74 };
  75
  76 static struct thread initial_thread;
  77 static struct thread *first_thread = &initial_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_fd = fd;
  84     thread->unix_pid  = 0;  /* not known yet */
  85     thread->mutex     = NULL;
  86     thread->debugger  = NULL;
  87     thread->wait      = NULL;
  88     thread->apc       = NULL;
  89     thread->apc_count = 0;
  90     thread->error     = 0;
  91     thread->state     = STARTING;
  92     thread->exit_code = 0x103;  /* STILL_ACTIVE */
  93     thread->next      = NULL;
  94     thread->prev      = NULL;
  95     thread->priority  = THREAD_PRIORITY_NORMAL;
  96     thread->affinity  = 1;
  97     thread->suspend   = 0;
  98 }
  99
 100 /* create the initial thread and start the main server loop */
 101 void create_initial_thread( int fd )
 102 {
 103     current = &initial_thread;
 104     init_thread( &initial_thread, fd );
 105     initial_thread.process = create_initial_process();
 106     add_process_thread( initial_thread.process, &initial_thread );
 107     add_client( fd, &initial_thread );
 108     grab_object( &initial_thread ); /* so that we never free it */
 109     select_loop();
 110 }
 111
 112 /* create a new thread */
 113 static struct thread *create_thread( int fd, void *pid, int suspend, int inherit, int *handle )
 114 {
 115     struct thread *thread;
 116     struct process *process;
 117
 118     if (!(thread = mem_alloc( sizeof(*thread) ))) return NULL;
 119
 120     if (!(process = get_process_from_id( pid )))
 121     {
 122         free( thread );
 123         return NULL;
 124     }
 125     init_thread( thread, fd );
 126     thread->process = process;
 127
 128     if (suspend) thread->suspend++;
 129
 130     thread->next = first_thread;
 131     first_thread->prev = thread;
 132     first_thread = thread;
 133     add_process_thread( process, thread );
 134
 135     if ((*handle = alloc_handle( current->process, thread,
 136                                  THREAD_ALL_ACCESS, inherit )) == -1) goto error;
 137     if (add_client( fd, thread ) == -1)
 138     {
 139         SET_ERROR( ERROR_TOO_MANY_OPEN_FILES );
 140         goto error;
 141     }
 142     return thread;
 143
 144  error:
 145     if (current) close_handle( current->process, *handle );
 146     remove_process_thread( process, thread );
 147     release_object( thread );
 148     return NULL;
 149 }
 150
 151 /* destroy a thread when its refcount is 0 */
 152 static void destroy_thread( struct object *obj )
 153 {
 154     struct thread *thread = (struct thread *)obj;
 155     assert( obj->ops == &thread_ops );
 156
 157     release_object( thread->process );
 158     if (thread->next) thread->next->prev = thread->prev;
 159     if (thread->prev) thread->prev->next = thread->next;
 160     else first_thread = thread->next;
 161     if (thread->apc) free( thread->apc );
 162     if (debug_level) memset( thread, 0xaa, sizeof(thread) );  /* catch errors */
 163     free( thread );
 164 }
 165
 166 /* dump a thread on stdout for debugging purposes */
 167 static void dump_thread( struct object *obj, int verbose )
 168 {
 169     struct thread *thread = (struct thread *)obj;
 170     assert( obj->ops == &thread_ops );
 171
 172     fprintf( stderr, "Thread pid=%d fd=%d\n",
 173              thread->unix_pid, thread->client_fd );
 174 }
 175
 176 static int thread_signaled( struct object *obj, struct thread *thread )
 177 {
 178     struct thread *mythread = (struct thread *)obj;
 179     return (mythread->state == TERMINATED);
 180 }
 181
 182 /* get a thread pointer from a thread id (and increment the refcount) */
 183 struct thread *get_thread_from_id( void *id )
 184 {
 185     struct thread *t = first_thread;
 186     while (t && (t != id)) t = t->next;
 187     if (t) grab_object( t );
 188     return t;
 189 }
 190
 191 /* get a thread from a handle (and increment the refcount) */
 192 struct thread *get_thread_from_handle( int handle, unsigned int access )
 193 {
 194     return (struct thread *)get_handle_obj( current->process, handle,
 195                                             access, &thread_ops );
 196 }
 197
 198 /* set all information about a thread */
 199 static void set_thread_info( struct thread *thread,
 200                              struct set_thread_info_request *req )
 201 {
 202     if (req->mask & SET_THREAD_INFO_PRIORITY)
 203         thread->priority = req->priority;
 204     if (req->mask & SET_THREAD_INFO_AFFINITY)
 205     {
 206         if (req->affinity != 1) SET_ERROR( ERROR_INVALID_PARAMETER );
 207         else thread->affinity = req->affinity;
 208     }
 209 }
 210
 211 /* suspend a thread */
 212 static int suspend_thread( struct thread *thread )
 213 {
 214     int old_count = thread->suspend;
 215     if (thread->suspend < MAXIMUM_SUSPEND_COUNT)
 216     {
 217         if (!thread->suspend++)
 218         {
 219             if (thread->unix_pid) kill( thread->unix_pid, SIGSTOP );
 220         }
 221     }
 222     return old_count;
 223 }
 224
 225 /* resume a thread */
 226 static int resume_thread( struct thread *thread )
 227 {
 228     int old_count = thread->suspend;
 229     if (thread->suspend > 0)
 230     {
 231         if (!--thread->suspend)
 232         {
 233             if (thread->unix_pid) kill( thread->unix_pid, SIGCONT );
 234         }
 235     }
 236     return old_count;
 237 }
 238
 239 /* suspend all threads but the current */
 240 void suspend_all_threads( void )
 241 {
 242     struct thread *thread;
 243     for ( thread = first_thread; thread; thread = thread->next )
 244         if ( thread != current )
 245             suspend_thread( thread );
 246 }
 247
 248 /* resume all threads but the current */
 249 void resume_all_threads( void )
 250 {
 251     struct thread *thread;
 252     for ( thread = first_thread; thread; thread = thread->next )
 253         if ( thread != current )
 254             resume_thread( thread );
 255 }
 256
 257 /* send a reply to a thread */
 258 int send_reply( struct thread *thread, int pass_fd, int n,
 259                 ... /* arg_1, len_1, ..., arg_n, len_n */ )
 260 {
 261     struct iovec vec[16];
 262     va_list args;
 263     int i;
 264
 265     assert( n < 16 );
 266     va_start( args, n );
 267     for (i = 0; i < n; i++)
 268     {
 269         vec[i].iov_base = va_arg( args, void * );
 270         vec[i].iov_len  = va_arg( args, int );
 271     }
 272     va_end( args );
 273     return send_reply_v( thread->client_fd, thread->error, pass_fd, vec, n );
 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->flags & SELECT_TIMEOUT) set_select_timeout( thread->client_fd, NULL );
 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     if (flags & SELECT_TIMEOUT)
 333     {
 334         gettimeofday( &wait->timeout, 0 );
 335         if (timeout)
 336         {
 337             wait->timeout.tv_usec += (timeout % 1000) * 1000;
 338             if (wait->timeout.tv_usec >= 1000000)
 339             {
 340                 wait->timeout.tv_usec -= 1000000;
 341                 wait->timeout.tv_sec++;
 342             }
 343             wait->timeout.tv_sec += timeout / 1000;
 344         }
 345     }
 346
 347     for (i = 0, entry = wait->queues; i < count; i++, entry++)
 348     {
 349         if (!(obj = get_handle_obj( thread->process, handles[i],
 350                                     SYNCHRONIZE, NULL )))
 351         {
 352             wait->count = i - 1;
 353             end_wait( thread );
 354             return 0;
 355         }
 356         entry->thread = thread;
 357         if (!obj->ops->add_queue( obj, entry ))
 358         {
 359             wait->count = i - 1;
 360             end_wait( thread );
 361             return 0;
 362         }
 363         release_object( obj );
 364     }
 365     return 1;
 366 }
 367
 368 /* check if the thread waiting condition is satisfied */
 369 static int check_wait( struct thread *thread, int *signaled )
 370 {
 371     int i;
 372     struct thread_wait *wait = thread->wait;
 373     struct wait_queue_entry *entry = wait->queues;
 374
 375     assert( wait );
 376     if (wait->flags & SELECT_ALL)
 377     {
 378         for (i = 0, entry = wait->queues; i < wait->count; i++, entry++)
 379             if (!entry->obj->ops->signaled( entry->obj, thread )) goto other_checks;
 380         /* Wait satisfied: tell it to all objects */
 381         *signaled = 0;
 382         for (i = 0, entry = wait->queues; i < wait->count; i++, entry++)
 383             if (entry->obj->ops->satisfied( entry->obj, thread ))
 384                 *signaled = STATUS_ABANDONED_WAIT_0;
 385         return 1;
 386     }
 387     else
 388     {
 389         for (i = 0, entry = wait->queues; i < wait->count; i++, entry++)
 390         {
 391             if (!entry->obj->ops->signaled( entry->obj, thread )) continue;
 392             /* Wait satisfied: tell it to the object */
 393             *signaled = i;
 394             if (entry->obj->ops->satisfied( entry->obj, thread ))
 395                 *signaled += STATUS_ABANDONED_WAIT_0;
 396             return 1;
 397         }
 398     }
 399
 400  other_checks:
 401     if ((wait->flags & SELECT_ALERTABLE) && thread->apc)
 402     {
 403         *signaled = STATUS_USER_APC;
 404         return 1;
 405     }
 406     if (wait->flags & SELECT_TIMEOUT)
 407     {
 408         struct timeval now;
 409         gettimeofday( &now, NULL );
 410         if ((now.tv_sec > wait->timeout.tv_sec) ||
 411             ((now.tv_sec == wait->timeout.tv_sec) &&
 412              (now.tv_usec >= wait->timeout.tv_usec)))
 413         {
 414             *signaled = STATUS_TIMEOUT;
 415             return 1;
 416         }
 417     }
 418     return 0;
 419 }
 420
 421 /* send the select reply to wake up the client */
 422 static void send_select_reply( struct thread *thread, int signaled )
 423 {
 424     struct select_reply reply;
 425     reply.signaled = signaled;
 426     if ((signaled == STATUS_USER_APC) && thread->apc)
 427     {
 428         struct thread_apc *apc = thread->apc;
 429         int len = thread->apc_count * sizeof(*apc);
 430         thread->apc = NULL;
 431         thread->apc_count = 0;
 432         send_reply( thread, -1, 2, &reply, sizeof(reply),
 433                     apc, len );
 434         free( apc );
 435     }
 436     else send_reply( thread, -1, 1, &reply, sizeof(reply) );
 437 }
 438
 439 /* attempt to wake up a thread */
 440 /* return 1 if OK, 0 if the wait condition is still not satisfied */
 441 static int wake_thread( struct thread *thread )
 442 {
 443     int signaled;
 444
 445     if (!check_wait( thread, &signaled )) return 0;
 446     end_wait( thread );
 447     send_select_reply( thread, signaled );
 448     return 1;
 449 }
 450
 451 /* sleep on a list of objects */
 452 static void sleep_on( struct thread *thread, int count, int *handles, int flags, int timeout )
 453 {
 454     assert( !thread->wait );
 455     if (!wait_on( thread, count, handles, flags, timeout ))
 456     {
 457         /* return an error */
 458         send_select_reply( thread, -1 );
 459         return;
 460     }
 461     if (!wake_thread( thread ))
 462     {
 463         /* we need to wait */
 464         if (flags & SELECT_TIMEOUT)
 465             set_select_timeout( thread->client_fd, &thread->wait->timeout );
 466     }
 467 }
 468
 469 /* timeout for the current thread */
 470 void thread_timeout(void)
 471 {
 472     assert( current->wait );
 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_fd, 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     if (current->suspend > 0) kill( current->unix_pid, SIGSTOP );
 562     reply.pid = current->process;
 563     reply.tid = current;
 564     send_reply( current, -1, 1, &reply, sizeof(reply) );
 565 }
 566
 567 /* terminate a thread */
 568 DECL_HANDLER(terminate_thread)
 569 {
 570     struct thread *thread;
 571
 572     if ((thread = get_thread_from_handle( req->handle, THREAD_TERMINATE )))
 573     {
 574         kill_thread( thread, req->exit_code );
 575         release_object( thread );
 576     }
 577     if (current) send_reply( current, -1, 0 );
 578 }
 579
 580 /* fetch information about a thread */
 581 DECL_HANDLER(get_thread_info)
 582 {
 583     struct thread *thread;
 584     struct get_thread_info_reply reply = { 0, 0, 0 };
 585
 586     if ((thread = get_thread_from_handle( req->handle, THREAD_QUERY_INFORMATION )))
 587     {
 588         reply.tid       = thread;
 589         reply.exit_code = thread->exit_code;
 590         reply.priority  = thread->priority;
 591         release_object( thread );
 592     }
 593     send_reply( current, -1, 1, &reply, sizeof(reply) );
 594 }
 595
 596 /* set information about a thread */
 597 DECL_HANDLER(set_thread_info)
 598 {
 599     struct thread *thread;
 600
 601     if ((thread = get_thread_from_handle( req->handle, THREAD_SET_INFORMATION )))
 602     {
 603         set_thread_info( thread, req );
 604         release_object( thread );
 605     }
 606     send_reply( current, -1, 0 );
 607 }
 608
 609 /* suspend a thread */
 610 DECL_HANDLER(suspend_thread)
 611 {
 612     struct thread *thread;
 613     struct suspend_thread_reply reply = { -1 };
 614     if ((thread = get_thread_from_handle( req->handle, THREAD_SUSPEND_RESUME )))
 615     {
 616         reply.count = suspend_thread( thread );
 617         release_object( thread );
 618     }
 619     send_reply( current, -1, 1, &reply, sizeof(reply) );
 620
 621 }
 622
 623 /* resume a thread */
 624 DECL_HANDLER(resume_thread)
 625 {
 626     struct thread *thread;
 627     struct resume_thread_reply reply = { -1 };
 628     if ((thread = get_thread_from_handle( req->handle, THREAD_SUSPEND_RESUME )))
 629     {
 630         reply.count = resume_thread( thread );
 631         release_object( thread );
 632     }
 633     send_reply( current, -1, 1, &reply, sizeof(reply) );
 634
 635 }
 636
 637 /* select on a handle list */
 638 DECL_HANDLER(select)
 639 {
 640     if (len != req->count * sizeof(int))
 641         fatal_protocol_error( "select: bad length %d for %d handles\n",
 642                               len, req->count );
 643     sleep_on( current, req->count, (int *)data, req->flags, req->timeout );
 644 }
 645
 646 /* queue an APC for a thread */
 647 DECL_HANDLER(queue_apc)
 648 {
 649     struct thread *thread;
 650     if ((thread = get_thread_from_handle( req->handle, THREAD_SET_CONTEXT )))
 651     {
 652         thread_queue_apc( thread, req->func, req->param );
 653         release_object( thread );
 654     }
 655     send_reply( current, -1, 0 );
 656 }