2 * Server-side thread management
4 * Copyright (C) 1998 Alexandre Julliard
15 #ifdef HAVE_SYS_MMAN_H
18 #include <sys/types.h>
19 #ifdef HAVE_SYS_SOCKET_H
20 # include <sys/socket.h>
38 struct wait_queue_entry
40 struct wait_queue_entry
*next
;
41 struct wait_queue_entry
*prev
;
43 struct thread
*thread
;
48 int count
; /* count of objects */
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];
56 /* asynchronous procedure calls */
60 void *func
; /* function to call in client */
61 void *param
; /* function param */
63 #define MAX_THREAD_APC 16 /* Max outstanding APCs for a thread */
66 /* thread operations */
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
);
73 static const struct object_ops thread_ops
=
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 */
86 no_get_file_info
, /* get_file_info */
87 destroy_thread
/* destroy */
90 static struct thread
*first_thread
;
92 /* allocate the buffer for the communication with the client */
93 static int alloc_client_buffer( struct thread
*thread
)
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
;
105 if (fd
!= -1) close( fd
);
109 /* create a new thread */
110 static struct thread
*create_thread( int fd
, struct process
*process
, int suspend
)
112 struct thread
*thread
;
115 int flags
= fcntl( fd
, F_GETFL
, 0 );
116 fcntl( fd
, F_SETFL
, flags
| O_NONBLOCK
);
118 if (!(thread
= alloc_object( &thread_ops
, fd
))) return NULL
;
120 thread
->unix_pid
= 0; /* not known yet */
122 thread
->mutex
= NULL
;
123 thread
->debug_ctx
= NULL
;
126 thread
->apc_count
= 0;
128 thread
->pass_fd
= -1;
129 thread
->state
= RUNNING
;
130 thread
->attached
= 0;
131 thread
->exit_code
= 0x103; /* STILL_ACTIVE */
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
;
140 if (!first_thread
) /* creating the first thread */
143 thread
->process
= process
= create_initial_process();
146 else thread
->process
= (struct process
*)grab_object( process
);
148 if ((thread
->next
= first_thread
) != NULL
) thread
->next
->prev
= thread
;
149 first_thread
= thread
;
150 add_process_thread( process
, thread
);
152 if ((buf_fd
= alloc_client_buffer( thread
)) == -1) goto error
;
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
);
160 remove_process_thread( process
, thread
);
161 release_object( thread
);
165 /* create the initial thread and start the main server loop */
166 void create_initial_thread( int fd
)
168 create_thread( fd
, NULL
, 0 );
172 /* handle a client event */
173 void thread_poll_event( struct object
*obj
, int event
)
175 struct thread
*thread
= (struct thread
*)obj
;
176 assert( obj
->ops
== &thread_ops
);
178 if (event
& (POLLERR
| POLLHUP
)) kill_thread( thread
, BROKEN_PIPE
);
181 if (event
& POLLOUT
) write_request( thread
);
182 if (event
& POLLIN
) read_request( thread
);
186 /* destroy a thread when its refcount is 0 */
187 static void destroy_thread( struct object
*obj
)
189 struct thread
*thread
= (struct thread
*)obj
;
190 assert( obj
->ops
== &thread_ops
);
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
);
202 /* dump a thread on stdout for debugging purposes */
203 static void dump_thread( struct object
*obj
, int verbose
)
205 struct thread
*thread
= (struct thread
*)obj
;
206 assert( obj
->ops
== &thread_ops
);
208 fprintf( stderr
, "Thread pid=%d teb=%p state=%d\n",
209 thread
->unix_pid
, thread
->teb
, thread
->state
);
212 static int thread_signaled( struct object
*obj
, struct thread
*thread
)
214 struct thread
*mythread
= (struct thread
*)obj
;
215 return (mythread
->state
== TERMINATED
);
218 /* get a thread pointer from a thread id (and increment the refcount) */
219 struct thread
*get_thread_from_id( void *id
)
221 struct thread
*t
= first_thread
;
222 while (t
&& (t
!= id
)) t
= t
->next
;
223 if (t
) grab_object( t
);
227 /* get a thread from a handle (and increment the refcount) */
228 struct thread
*get_thread_from_handle( int handle
, unsigned int access
)
230 return (struct thread
*)get_handle_obj( current
->process
, handle
,
231 access
, &thread_ops
);
234 /* find a thread from a Unix pid */
235 struct thread
*get_thread_from_pid( int pid
)
237 struct thread
*t
= first_thread
;
238 while (t
&& (t
->unix_pid
!= pid
)) t
= t
->next
;
242 /* set all information about a thread */
243 static void set_thread_info( struct thread
*thread
,
244 struct set_thread_info_request
*req
)
246 if (req
->mask
& SET_THREAD_INFO_PRIORITY
)
247 thread
->priority
= req
->priority
;
248 if (req
->mask
& SET_THREAD_INFO_AFFINITY
)
250 if (req
->affinity
!= 1) set_error( ERROR_INVALID_PARAMETER
);
251 else thread
->affinity
= req
->affinity
;
255 /* suspend a thread */
256 int suspend_thread( struct thread
*thread
, int check_limit
)
258 int old_count
= thread
->suspend
;
259 if (thread
->suspend
< MAXIMUM_SUSPEND_COUNT
|| !check_limit
)
261 if (!(thread
->process
->suspend
+ thread
->suspend
++)) stop_thread( thread
);
263 else set_error( ERROR_SIGNAL_REFUSED
);
267 /* resume a thread */
268 int resume_thread( struct thread
*thread
)
270 int old_count
= thread
->suspend
;
271 if (thread
->suspend
> 0)
273 if (!(--thread
->suspend
+ thread
->process
->suspend
)) continue_thread( thread
);
278 /* suspend all threads but the current */
279 void suspend_all_threads( void )
281 struct thread
*thread
;
282 for ( thread
= first_thread
; thread
; thread
= thread
->next
)
283 if ( thread
!= current
)
284 suspend_thread( thread
, 0 );
287 /* resume all threads but the current */
288 void resume_all_threads( void )
290 struct thread
*thread
;
291 for ( thread
= first_thread
; thread
; thread
= thread
->next
)
292 if ( thread
!= current
)
293 resume_thread( thread
);
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
)
301 entry
->prev
= obj
->tail
;
303 if (obj
->tail
) obj
->tail
->next
= entry
;
304 else obj
->head
= entry
;
309 /* remove a thread from an object wait queue */
310 void remove_queue( struct object
*obj
, struct wait_queue_entry
*entry
)
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
);
320 static void end_wait( struct thread
*thread
)
322 struct thread_wait
*wait
= thread
->wait
;
323 struct wait_queue_entry
*entry
;
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
);
334 /* build the thread wait structure */
335 static int wait_on( int count
, struct object
*objects
[], int flags
,
336 int timeout
, sleep_reply func
)
338 struct thread_wait
*wait
;
339 struct wait_queue_entry
*entry
;
342 if (!(wait
= mem_alloc( sizeof(*wait
) + (count
-1) * sizeof(*entry
) ))) return 0;
343 current
->wait
= wait
;
348 if (flags
& SELECT_TIMEOUT
)
350 gettimeofday( &wait
->timeout
, 0 );
351 add_timeout( &wait
->timeout
, timeout
);
354 for (i
= 0, entry
= wait
->queues
; i
< count
; i
++, entry
++)
356 struct object
*obj
= objects
[i
];
357 entry
->thread
= current
;
358 if (!obj
->ops
->add_queue( obj
, entry
))
368 /* check if the thread waiting condition is satisfied */
369 static int check_wait( struct thread
*thread
, struct object
**object
)
372 struct thread_wait
*wait
= thread
->wait
;
373 struct wait_queue_entry
*entry
= wait
->queues
;
377 if (wait
->flags
& SELECT_ALL
)
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 */
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
;
394 for (i
= 0, entry
= wait
->queues
; i
< wait
->count
; i
++, entry
++)
396 if (!entry
->obj
->ops
->signaled( entry
->obj
, thread
)) continue;
397 /* Wait satisfied: tell it to the object */
399 *object
= entry
->obj
;
400 if (entry
->obj
->ops
->satisfied( entry
->obj
, thread
))
401 signaled
= i
+ STATUS_ABANDONED_WAIT_0
;
407 if ((wait
->flags
& SELECT_ALERTABLE
) && thread
->apc
) return STATUS_USER_APC
;
408 if (wait
->flags
& SELECT_TIMEOUT
)
411 gettimeofday( &now
, NULL
);
412 if (!time_before( &now
, &wait
->timeout
)) return STATUS_TIMEOUT
;
417 /* build a reply to the select request */
418 static void build_select_reply( struct thread
*thread
, struct object
*obj
, int signaled
)
420 struct select_request
*req
= get_req_ptr( thread
);
421 req
->signaled
= signaled
;
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
)
429 struct object
*object
;
430 if ((signaled
= check_wait( thread
, &object
)) == -1) return 0;
432 thread
->wait
->reply( thread
, object
, signaled
);
437 /* thread wait timeout */
438 static void thread_timeout( void *ptr
)
440 struct thread
*thread
= ptr
;
441 if (debug_level
) fprintf( stderr
, "%08x: *timeout*\n", (unsigned int)thread
);
442 assert( thread
->wait
);
444 thread
->wait
->user
= NULL
;
445 thread
->wait
->reply( thread
, NULL
, STATUS_TIMEOUT
);
447 send_reply( thread
);
450 /* sleep on a list of objects */
451 int sleep_on( int count
, struct object
*objects
[], int flags
, int timeout
, sleep_reply func
)
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
)
459 if (!(current
->wait
->user
= add_timeout_user( ¤t
->wait
->timeout
,
460 thread_timeout
, current
)))
469 /* select on a list of handles */
470 static int select_on( int count
, int *handles
, int flags
, int timeout
)
474 struct object
*objects
[MAXIMUM_WAIT_OBJECTS
];
476 if ((count
< 0) || (count
> MAXIMUM_WAIT_OBJECTS
))
478 set_error( ERROR_INVALID_PARAMETER
);
481 for (i
= 0; i
< count
; i
++)
483 if (!(objects
[i
] = get_handle_obj( current
->process
, handles
[i
], SYNCHRONIZE
, NULL
)))
486 if (i
== count
) ret
= sleep_on( count
, objects
, flags
, timeout
, build_select_reply
);
487 while (--i
>= 0) release_object( objects
[i
] );
491 /* attempt to wake threads sleeping on the object wait queue */
492 void wake_up( struct object
*obj
, int max
)
494 struct wait_queue_entry
*entry
= obj
->head
;
498 struct thread
*thread
= entry
->thread
;
500 if (wake_thread( thread
))
502 send_reply( thread
);
503 if (max
&& !--max
) break;
508 /* queue an async procedure call */
509 static int thread_queue_apc( struct thread
*thread
, void *func
, void *param
)
511 struct thread_apc
*apc
;
514 if (!(thread
->apc
= mem_alloc( MAX_THREAD_APC
* sizeof(*apc
) )))
516 thread
->apc_count
= 0;
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
;
524 if (wake_thread( thread
)) send_reply( thread
);
529 /* kill a thread on the spot */
530 void kill_thread( struct thread
*thread
, int exit_code
)
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
);
547 /* create a new thread */
548 DECL_HANDLER(new_thread
)
550 struct thread
*thread
;
551 struct process
*process
;
554 if (!(process
= get_process_from_id( req
->pid
))) return;
556 if (socketpair( AF_UNIX
, SOCK_STREAM
, 0, sock
) != -1)
558 if ((thread
= create_thread( sock
[0], process
, req
->suspend
)))
561 if ((req
->handle
= alloc_handle( current
->process
, thread
,
562 THREAD_ALL_ACCESS
, req
->inherit
)) != -1)
564 set_reply_fd( current
, sock
[1] );
565 release_object( process
);
566 /* thread object will be released when the thread gets killed */
569 release_object( thread
);
573 else file_set_error();
574 release_object( process
);
577 /* retrieve the thread buffer file descriptor */
578 DECL_HANDLER(get_thread_buffer
)
580 fatal_protocol_error( current
, "get_thread_buffer: should never get called directly\n" );
583 /* initialize a new thread */
584 DECL_HANDLER(init_thread
)
586 if (current
->unix_pid
)
588 fatal_protocol_error( current
, "init_thread: already running\n" );
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
;
598 /* terminate a thread */
599 DECL_HANDLER(terminate_thread
)
601 struct thread
*thread
;
603 if ((thread
= get_thread_from_handle( req
->handle
, THREAD_TERMINATE
)))
605 kill_thread( thread
, req
->exit_code
);
606 release_object( thread
);
610 /* fetch information about a thread */
611 DECL_HANDLER(get_thread_info
)
613 struct thread
*thread
;
615 if ((thread
= get_thread_from_handle( req
->handle
, THREAD_QUERY_INFORMATION
)))
618 req
->exit_code
= thread
->exit_code
;
619 req
->priority
= thread
->priority
;
620 release_object( thread
);
624 /* set information about a thread */
625 DECL_HANDLER(set_thread_info
)
627 struct thread
*thread
;
629 if ((thread
= get_thread_from_handle( req
->handle
, THREAD_SET_INFORMATION
)))
631 set_thread_info( thread
, req
);
632 release_object( thread
);
636 /* suspend a thread */
637 DECL_HANDLER(suspend_thread
)
639 struct thread
*thread
;
641 if ((thread
= get_thread_from_handle( req
->handle
, THREAD_SUSPEND_RESUME
)))
643 req
->count
= suspend_thread( thread
, 1 );
644 release_object( thread
);
648 /* resume a thread */
649 DECL_HANDLER(resume_thread
)
651 struct thread
*thread
;
653 if ((thread
= get_thread_from_handle( req
->handle
, THREAD_SUSPEND_RESUME
)))
655 req
->count
= resume_thread( thread
);
656 release_object( thread
);
660 /* select on a handle list */
663 if (!select_on( req
->count
, req
->handles
, req
->flags
, req
->timeout
))
667 /* queue an APC for a thread */
668 DECL_HANDLER(queue_apc
)
670 struct thread
*thread
;
671 if ((thread
= get_thread_from_handle( req
->handle
, THREAD_SET_CONTEXT
)))
673 thread_queue_apc( thread
, req
->func
, req
->param
);
674 release_object( thread
);
678 /* get list of APC to call */
679 DECL_HANDLER(get_apcs
)
681 if ((req
->count
= current
->apc_count
))
683 memcpy( req
->apcs
, current
->apc
, current
->apc_count
* sizeof(*current
->apc
) );
684 free( current
->apc
);
686 current
->apc_count
= 0;