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