2 * Server-side thread management
4 * Copyright (C) 1998 Alexandre Julliard
13 #include <sys/types.h>
20 #include "server/thread.h"
25 struct wait_queue_entry
27 struct wait_queue_entry
*next
;
28 struct wait_queue_entry
*prev
;
30 struct thread
*thread
;
35 int count
; /* count of objects */
37 struct timeval timeout
;
38 struct wait_queue_entry queues
[1];
41 /* asynchronous procedure calls */
45 void *func
; /* function to call in client */
46 void *param
; /* function param */
48 #define MAX_THREAD_APC 16 /* Max outstanding APCs for a thread */
51 /* thread operations */
53 static void dump_thread( struct object
*obj
, int verbose
);
54 static int thread_signaled( struct object
*obj
, struct thread
*thread
);
55 static void destroy_thread( struct object
*obj
);
57 static const struct object_ops thread_ops
=
71 static struct thread
*first_thread
;
74 /* create a new thread */
75 struct thread
*create_thread( int fd
, void *pid
, int *thread_handle
,
78 struct thread
*thread
;
79 struct process
*process
;
81 if (!(thread
= mem_alloc( sizeof(*thread
) ))) return NULL
;
83 if (pid
) process
= get_process_from_id( pid
);
84 else process
= create_process();
91 init_object( &thread
->obj
, &thread_ops
, NULL
);
92 thread
->client_fd
= fd
;
93 thread
->process
= process
;
94 thread
->unix_pid
= 0; /* not known yet */
99 thread
->apc_count
= 0;
101 thread
->state
= STARTING
;
102 thread
->exit_code
= 0x103; /* STILL_ACTIVE */
103 thread
->next
= first_thread
;
105 thread
->priority
= THREAD_PRIORITY_NORMAL
;
106 thread
->affinity
= 1;
109 if (first_thread
) first_thread
->prev
= thread
;
110 first_thread
= thread
;
111 add_process_thread( process
, thread
);
113 *thread_handle
= *process_handle
= -1;
116 if ((*thread_handle
= alloc_handle( current
->process
, thread
,
117 THREAD_ALL_ACCESS
, 0 )) == -1)
122 if ((*process_handle
= alloc_handle( current
->process
, process
,
123 PROCESS_ALL_ACCESS
, 0 )) == -1)
127 if (add_client( fd
, thread
) == -1) goto error
;
134 close_handle( current
->process
, *thread_handle
);
135 close_handle( current
->process
, *process_handle
);
137 remove_process_thread( process
, thread
);
138 release_object( thread
);
142 /* destroy a thread when its refcount is 0 */
143 static void destroy_thread( struct object
*obj
)
145 struct thread
*thread
= (struct thread
*)obj
;
146 assert( obj
->ops
== &thread_ops
);
148 release_object( thread
->process
);
149 if (thread
->next
) thread
->next
->prev
= thread
->prev
;
150 if (thread
->prev
) thread
->prev
->next
= thread
->next
;
151 else first_thread
= thread
->next
;
152 if (thread
->name
) free( thread
->name
);
153 if (thread
->apc
) free( thread
->apc
);
154 if (debug_level
) memset( thread
, 0xaa, sizeof(thread
) ); /* catch errors */
158 /* dump a thread on stdout for debugging purposes */
159 static void dump_thread( struct object
*obj
, int verbose
)
161 struct thread
*thread
= (struct thread
*)obj
;
162 assert( obj
->ops
== &thread_ops
);
164 fprintf( stderr
, "Thread pid=%d fd=%d name='%s'\n",
165 thread
->unix_pid
, thread
->client_fd
, thread
->name
);
168 static int thread_signaled( struct object
*obj
, struct thread
*thread
)
170 struct thread
*mythread
= (struct thread
*)obj
;
171 return (mythread
->state
== TERMINATED
);
174 /* get a thread pointer from a thread id (and increment the refcount) */
175 struct thread
*get_thread_from_id( void *id
)
177 struct thread
*t
= first_thread
;
178 while (t
&& (t
!= id
)) t
= t
->next
;
179 if (t
) grab_object( t
);
183 /* get a thread from a handle (and increment the refcount) */
184 struct thread
*get_thread_from_handle( int handle
, unsigned int access
)
186 return (struct thread
*)get_handle_obj( current
->process
, handle
,
187 access
, &thread_ops
);
190 /* get all information about a thread */
191 void get_thread_info( struct thread
*thread
,
192 struct get_thread_info_reply
*reply
)
195 reply
->exit_code
= thread
->exit_code
;
196 reply
->priority
= thread
->priority
;
200 /* set all information about a thread */
201 void set_thread_info( struct thread
*thread
,
202 struct set_thread_info_request
*req
)
204 if (req
->mask
& SET_THREAD_INFO_PRIORITY
)
205 thread
->priority
= req
->priority
;
206 if (req
->mask
& SET_THREAD_INFO_AFFINITY
)
208 if (req
->affinity
!= 1) SET_ERROR( ERROR_INVALID_PARAMETER
);
209 else thread
->affinity
= req
->affinity
;
213 /* suspend a thread */
214 int suspend_thread( struct thread
*thread
)
216 int old_count
= thread
->suspend
;
217 if (thread
->suspend
< MAXIMUM_SUSPEND_COUNT
)
219 if (!thread
->suspend
++)
221 if (thread
->unix_pid
) kill( thread
->unix_pid
, SIGSTOP
);
227 /* resume a thread */
228 int resume_thread( struct thread
*thread
)
230 int old_count
= thread
->suspend
;
231 if (thread
->suspend
> 0)
233 if (!--thread
->suspend
)
235 if (thread
->unix_pid
) kill( thread
->unix_pid
, SIGCONT
);
241 /* send a reply to a thread */
242 int send_reply( struct thread
*thread
, int pass_fd
, int n
,
243 ... /* arg_1, len_1, ..., arg_n, len_n */ )
245 struct iovec vec
[16];
251 for (i
= 0; i
< n
; i
++)
253 vec
[i
].iov_base
= va_arg( args
, void * );
254 vec
[i
].iov_len
= va_arg( args
, int );
257 return send_reply_v( thread
->client_fd
, thread
->error
, pass_fd
, vec
, n
);
260 /* add a thread to an object wait queue; return 1 if OK, 0 on error */
261 int add_queue( struct object
*obj
, struct wait_queue_entry
*entry
)
265 entry
->prev
= obj
->tail
;
267 if (obj
->tail
) obj
->tail
->next
= entry
;
268 else obj
->head
= entry
;
273 /* remove a thread from an object wait queue */
274 void remove_queue( struct object
*obj
, struct wait_queue_entry
*entry
)
276 if (entry
->next
) entry
->next
->prev
= entry
->prev
;
277 else obj
->tail
= entry
->prev
;
278 if (entry
->prev
) entry
->prev
->next
= entry
->next
;
279 else obj
->head
= entry
->next
;
280 release_object( obj
);
284 static void end_wait( struct thread
*thread
)
286 struct thread_wait
*wait
= thread
->wait
;
287 struct wait_queue_entry
*entry
;
291 for (i
= 0, entry
= wait
->queues
; i
< wait
->count
; i
++, entry
++)
292 entry
->obj
->ops
->remove_queue( entry
->obj
, entry
);
293 if (wait
->flags
& SELECT_TIMEOUT
) set_select_timeout( thread
->client_fd
, NULL
);
298 /* build the thread wait structure */
299 static int wait_on( struct thread
*thread
, int count
,
300 int *handles
, int flags
, int timeout
)
302 struct thread_wait
*wait
;
303 struct wait_queue_entry
*entry
;
307 if ((count
< 0) || (count
> MAXIMUM_WAIT_OBJECTS
))
309 SET_ERROR( ERROR_INVALID_PARAMETER
);
312 if (!(wait
= mem_alloc( sizeof(*wait
) + (count
-1) * sizeof(*entry
) ))) return 0;
316 if (flags
& SELECT_TIMEOUT
)
318 gettimeofday( &wait
->timeout
, 0 );
321 wait
->timeout
.tv_usec
+= (timeout
% 1000) * 1000;
322 if (wait
->timeout
.tv_usec
>= 1000000)
324 wait
->timeout
.tv_usec
-= 1000000;
325 wait
->timeout
.tv_sec
++;
327 wait
->timeout
.tv_sec
+= timeout
/ 1000;
331 for (i
= 0, entry
= wait
->queues
; i
< count
; i
++, entry
++)
333 if (!(obj
= get_handle_obj( thread
->process
, handles
[i
],
334 SYNCHRONIZE
, NULL
)))
340 entry
->thread
= thread
;
341 if (!obj
->ops
->add_queue( obj
, entry
))
347 release_object( obj
);
352 /* check if the thread waiting condition is satisfied */
353 static int check_wait( struct thread
*thread
, int *signaled
)
356 struct thread_wait
*wait
= thread
->wait
;
357 struct wait_queue_entry
*entry
= wait
->queues
;
360 if (wait
->flags
& SELECT_ALL
)
362 for (i
= 0, entry
= wait
->queues
; i
< wait
->count
; i
++, entry
++)
363 if (!entry
->obj
->ops
->signaled( entry
->obj
, thread
)) goto other_checks
;
364 /* Wait satisfied: tell it to all objects */
366 for (i
= 0, entry
= wait
->queues
; i
< wait
->count
; i
++, entry
++)
367 if (entry
->obj
->ops
->satisfied( entry
->obj
, thread
))
368 *signaled
= STATUS_ABANDONED_WAIT_0
;
373 for (i
= 0, entry
= wait
->queues
; i
< wait
->count
; i
++, entry
++)
375 if (!entry
->obj
->ops
->signaled( entry
->obj
, thread
)) continue;
376 /* Wait satisfied: tell it to the object */
378 if (entry
->obj
->ops
->satisfied( entry
->obj
, thread
))
379 *signaled
+= STATUS_ABANDONED_WAIT_0
;
385 if ((wait
->flags
& SELECT_ALERTABLE
) && thread
->apc
)
387 *signaled
= STATUS_USER_APC
;
390 if (wait
->flags
& SELECT_TIMEOUT
)
393 gettimeofday( &now
, NULL
);
394 if ((now
.tv_sec
> wait
->timeout
.tv_sec
) ||
395 ((now
.tv_sec
== wait
->timeout
.tv_sec
) &&
396 (now
.tv_usec
>= wait
->timeout
.tv_usec
)))
398 *signaled
= STATUS_TIMEOUT
;
405 /* send the select reply to wake up the client */
406 static void send_select_reply( struct thread
*thread
, int signaled
)
408 struct select_reply reply
;
409 reply
.signaled
= signaled
;
410 if ((signaled
== STATUS_USER_APC
) && thread
->apc
)
412 struct thread_apc
*apc
= thread
->apc
;
413 int len
= thread
->apc_count
* sizeof(*apc
);
415 thread
->apc_count
= 0;
416 send_reply( thread
, -1, 2, &reply
, sizeof(reply
),
420 else send_reply( thread
, -1, 1, &reply
, sizeof(reply
) );
423 /* attempt to wake up a thread */
424 /* return 1 if OK, 0 if the wait condition is still not satisfied */
425 static int wake_thread( struct thread
*thread
)
429 if (!check_wait( thread
, &signaled
)) return 0;
431 send_select_reply( thread
, signaled
);
435 /* sleep on a list of objects */
436 void sleep_on( struct thread
*thread
, int count
, int *handles
, int flags
, int timeout
)
438 assert( !thread
->wait
);
439 if (!wait_on( thread
, count
, handles
, flags
, timeout
))
441 /* return an error */
442 send_select_reply( thread
, -1 );
445 if (!wake_thread( thread
))
447 /* we need to wait */
448 if (flags
& SELECT_TIMEOUT
)
449 set_select_timeout( thread
->client_fd
, &thread
->wait
->timeout
);
453 /* timeout for the current thread */
454 void thread_timeout(void)
456 assert( current
->wait
);
458 send_select_reply( current
, STATUS_TIMEOUT
);
461 /* attempt to wake threads sleeping on the object wait queue */
462 void wake_up( struct object
*obj
, int max
)
464 struct wait_queue_entry
*entry
= obj
->head
;
468 struct wait_queue_entry
*next
= entry
->next
;
469 if (wake_thread( entry
->thread
))
471 if (max
&& !--max
) break;
477 /* queue an async procedure call */
478 int thread_queue_apc( struct thread
*thread
, void *func
, void *param
)
480 struct thread_apc
*apc
;
483 SET_ERROR( ERROR_INVALID_PARAMETER
);
488 if (!(thread
->apc
= mem_alloc( MAX_THREAD_APC
* sizeof(*apc
) )))
490 thread
->apc_count
= 0;
492 else if (thread
->apc_count
>= MAX_THREAD_APC
) return 0;
493 thread
->apc
[thread
->apc_count
].func
= func
;
494 thread
->apc
[thread
->apc_count
].param
= param
;
496 wake_thread( thread
);
500 /* kill a thread on the spot */
501 void kill_thread( struct thread
*thread
, int exit_code
)
503 if (thread
->state
== TERMINATED
) return; /* already killed */
504 if (thread
->unix_pid
) kill( thread
->unix_pid
, SIGTERM
);
505 remove_client( thread
->client_fd
, exit_code
); /* this will call thread_killed */
508 /* a thread has been killed */
509 void thread_killed( struct thread
*thread
, int exit_code
)
511 thread
->state
= TERMINATED
;
512 thread
->exit_code
= exit_code
;
513 if (thread
->wait
) end_wait( thread
);
514 abandon_mutexes( thread
);
515 remove_process_thread( thread
->process
, thread
);
516 wake_up( &thread
->obj
, 0 );
517 release_object( thread
);