2 * threadpool.c: global thread pool
5 * Dietmar Maurer (dietmar@ximian.com)
6 * Gonzalo Paniagua Javier (gonzalo@ximian.com)
8 * (C) 2001-2003 Ximian, Inc.
9 * (c) 2004,2005 Novell, Inc. (http://www.novell.com)
17 #define _WIN32_WINNT 0x0500
20 #define THREADS_PER_CPU 5 /* 20 + THREADS_PER_CPU * number of CPUs */
22 #include <mono/metadata/domain-internals.h>
23 #include <mono/metadata/tabledefs.h>
24 #include <mono/metadata/threads.h>
25 #include <mono/metadata/threads-types.h>
26 #include <mono/metadata/threadpool-internals.h>
27 #include <mono/metadata/exception.h>
28 #include <mono/metadata/file-io.h>
29 #include <mono/metadata/monitor.h>
30 #include <mono/metadata/mono-mlist.h>
31 #include <mono/metadata/marshal.h>
32 #include <mono/metadata/socket-io.h>
33 #include <mono/io-layer/io-layer.h>
34 #include <mono/os/gc_wrapper.h>
36 #ifdef HAVE_SYS_TIME_H
39 #include <sys/types.h>
46 #include <mono/utils/mono-poll.h>
48 #include <sys/epoll.h>
51 #include "mono/io-layer/socket-wrappers.h"
53 #include "threadpool.h"
55 #define THREAD_WANTS_A_BREAK(t) ((t->state & (ThreadState_StopRequested | \
56 ThreadState_SuspendRequested)) != 0)
60 /* maximum number of worker threads */
61 static int mono_max_worker_threads;
62 static int mono_min_worker_threads;
63 static int mono_io_max_worker_threads;
64 static int mono_io_min_worker_threads;
66 /* current number of worker threads */
67 static int mono_worker_threads = 0;
68 static int io_worker_threads = 0;
70 /* current number of busy threads */
71 static int busy_worker_threads = 0;
72 static int busy_io_worker_threads;
74 /* mono_thread_pool_init called */
77 /* we use this to store a reference to the AsyncResult to avoid GC */
78 static MonoGHashTable *ares_htable = NULL;
80 static CRITICAL_SECTION ares_lock;
81 static CRITICAL_SECTION io_queue_lock;
82 static int pending_io_items;
85 CRITICAL_SECTION io_lock; /* access to sock_to_state */
88 MonoGHashTable *sock_to_state;
90 HANDLE new_sem; /* access to newpfd and write side of the pipe */
92 gboolean epoll_disabled;
98 static SocketIOData socket_io_data;
100 /* we append a job */
101 static HANDLE job_added;
102 static HANDLE io_job_added;
104 /* Keep in sync with the System.MonoAsyncCall class which provides GC tracking */
107 MonoMethodMessage *msg;
108 MonoMethod *cb_method;
109 MonoDelegate *cb_target;
113 /* This is a HANDLE, we use guint64 so the managed object layout remains constant */
123 static void async_invoke_thread (gpointer data);
124 static void append_job (CRITICAL_SECTION *cs, TPQueue *list, MonoObject *ar);
125 static void start_thread_or_queue (MonoAsyncResult *ares);
126 static void mono_async_invoke (MonoAsyncResult *ares);
127 static MonoObject* dequeue_job (CRITICAL_SECTION *cs, TPQueue *list);
128 static void free_queue (TPQueue *list);
130 static TPQueue async_call_queue = {NULL, 0, 0};
131 static TPQueue async_io_queue = {NULL, 0, 0};
133 static MonoClass *async_call_klass;
134 static MonoClass *socket_async_call_klass;
135 static MonoClass *process_async_call_klass;
137 #define INIT_POLLFD(a, b, c) {(a)->fd = b; (a)->events = c; (a)->revents = 0;}
146 AIO_OP_RECV_JUST_CALLBACK,
147 AIO_OP_SEND_JUST_CALLBACK,
153 socket_io_cleanup (SocketIOData *data)
157 if (data->inited == 0)
160 EnterCriticalSection (&data->io_lock);
162 #ifdef PLATFORM_WIN32
163 closesocket (data->pipe [0]);
164 closesocket (data->pipe [1]);
166 close (data->pipe [0]);
167 close (data->pipe [1]);
172 CloseHandle (data->new_sem);
173 data->new_sem = NULL;
174 mono_g_hash_table_destroy (data->sock_to_state);
175 data->sock_to_state = NULL;
176 free_queue (&async_io_queue);
177 release = (gint) InterlockedCompareExchange (&io_worker_threads, 0, -1);
179 ReleaseSemaphore (io_job_added, release, NULL);
180 g_free (data->newpfd);
183 if (FALSE == data->epoll_disabled)
184 close (data->epollfd);
186 LeaveCriticalSection (&data->io_lock);
190 get_event_from_state (MonoSocketAsyncResult *state)
192 switch (state->operation) {
195 case AIO_OP_RECV_JUST_CALLBACK:
196 case AIO_OP_RECEIVEFROM:
197 case AIO_OP_READPIPE:
200 case AIO_OP_SEND_JUST_CALLBACK:
204 default: /* Should never happen */
205 g_print ("get_event_from_state: unknown value in switch!!!\n");
211 get_events_from_list (MonoMList *list)
213 MonoSocketAsyncResult *state;
216 while (list && (state = (MonoSocketAsyncResult *)mono_mlist_get_data (list))) {
217 events |= get_event_from_state (state);
218 list = mono_mlist_next (list);
224 #define ICALL_RECV(x) ves_icall_System_Net_Sockets_Socket_Receive_internal (\
225 (SOCKET) x->handle, x->buffer, x->offset, x->size,\
226 x->socket_flags, &x->error);
228 #define ICALL_SEND(x) ves_icall_System_Net_Sockets_Socket_Send_internal (\
229 (SOCKET) x->handle, x->buffer, x->offset, x->size,\
230 x->socket_flags, &x->error);
233 async_invoke_io_thread (gpointer data)
237 thread = mono_thread_current ();
238 thread->threadpool_thread = TRUE;
239 ves_icall_System_Threading_Thread_SetState (thread, ThreadState_Background);
242 MonoSocketAsyncResult *state;
245 state = (MonoSocketAsyncResult *) data;
247 InterlockedDecrement (&pending_io_items);
249 switch (state->operation) {
251 state->total = ICALL_RECV (state);
254 state->total = ICALL_SEND (state);
258 /* worker threads invokes methods in different domains,
259 * so we need to set the right domain here */
260 domain = ((MonoObject *)ar)->vtable->domain;
261 mono_thread_push_appdomain_ref (domain);
262 if (mono_domain_set (domain, FALSE)) {
265 mono_async_invoke (ar);
266 ac = (ASyncCall *) ar->object_data;
268 if (ac->msg->exc != NULL)
269 mono_unhandled_exception (ac->msg->exc);
271 mono_domain_set (mono_get_root_domain (), TRUE);
273 mono_thread_pop_appdomain_ref ();
274 InterlockedDecrement (&busy_io_worker_threads);
277 data = dequeue_job (&io_queue_lock, &async_io_queue);
282 guint32 start_time = GetTickCount ();
285 wr = WaitForSingleObjectEx (io_job_added, (guint32)timeout, TRUE);
286 if (THREAD_WANTS_A_BREAK (thread))
287 mono_thread_interruption_checkpoint ();
289 timeout -= GetTickCount () - start_time;
291 if (wr != WAIT_TIMEOUT)
292 data = dequeue_job (&io_queue_lock, &async_io_queue);
294 while (!data && timeout > 0);
298 if (InterlockedDecrement (&io_worker_threads) < 2) {
299 /* If we have pending items, keep the thread alive */
300 if (InterlockedCompareExchange (&pending_io_items, 0, 0) != 0) {
301 InterlockedIncrement (&io_worker_threads);
308 InterlockedIncrement (&busy_io_worker_threads);
311 g_assert_not_reached ();
315 start_io_thread_or_queue (MonoSocketAsyncResult *ares)
320 busy = (int) InterlockedCompareExchange (&busy_io_worker_threads, 0, -1);
321 worker = (int) InterlockedCompareExchange (&io_worker_threads, 0, -1);
322 if (worker <= ++busy &&
323 worker < mono_io_max_worker_threads) {
324 InterlockedIncrement (&busy_io_worker_threads);
325 InterlockedIncrement (&io_worker_threads);
326 domain = ((ares) ? ((MonoObject *) ares)->vtable->domain : mono_domain_get ());
327 mono_thread_create (mono_get_root_domain (), async_invoke_io_thread, ares);
329 append_job (&io_queue_lock, &async_io_queue, (MonoObject*)ares);
330 ReleaseSemaphore (io_job_added, 1, NULL);
335 process_io_event (MonoMList *list, int event)
337 MonoSocketAsyncResult *state;
343 state = (MonoSocketAsyncResult *) mono_mlist_get_data (list);
344 if (get_event_from_state (state) == event)
347 list = mono_mlist_next (list);
351 oldlist = mono_mlist_remove_item (oldlist, list);
353 g_print ("Dispatching event %d on socket %d\n", event, state->handle);
355 InterlockedIncrement (&pending_io_items);
356 start_io_thread_or_queue (state);
363 mark_bad_fds (mono_pollfd *pfds, int nfds)
369 for (i = 0; i < nfds; i++) {
374 ret = mono_poll (pfd, 1, 0);
375 if (ret == -1 && errno == EBADF) {
376 pfd->revents |= MONO_POLLNVAL;
378 } else if (ret == 1) {
387 socket_io_poll_main (gpointer p)
389 #define INITIAL_POLLFD_SIZE 1024
390 #define POLL_ERRORS (MONO_POLLERR | MONO_POLLHUP | MONO_POLLNVAL)
391 SocketIOData *data = p;
398 thread = mono_thread_current ();
399 thread->threadpool_thread = TRUE;
400 ves_icall_System_Threading_Thread_SetState (thread, ThreadState_Background);
402 allocated = INITIAL_POLLFD_SIZE;
403 pfds = g_new0 (mono_pollfd, allocated);
404 INIT_POLLFD (pfds, data->pipe [0], MONO_POLLIN);
405 for (i = 1; i < allocated; i++)
406 INIT_POLLFD (&pfds [i], -1, 0);
416 if (THREAD_WANTS_A_BREAK (thread))
417 mono_thread_interruption_checkpoint ();
420 nsock = mono_poll (pfds, maxfd, -1);
421 } while (nsock == -1 && errno == EINTR);
424 * Apart from EINTR, we only check EBADF, for the rest:
425 * EINVAL: mono_poll() 'protects' us from descriptor
426 * numbers above the limit if using select() by marking
427 * then as MONO_POLLERR. If a system poll() is being
428 * used, the number of descriptor we're passing will not
429 * be over sysconf(_SC_OPEN_MAX), as the error would have
430 * happened when opening.
432 * EFAULT: we own the memory pointed by pfds.
433 * ENOMEM: we're doomed anyway
437 if (nsock == -1 && errno == EBADF) {
438 pfds->revents = 0; /* Just in case... */
439 nsock = mark_bad_fds (pfds, maxfd);
442 if ((pfds->revents & POLL_ERRORS) != 0) {
443 /* We're supposed to die now, as the pipe has been closed */
445 socket_io_cleanup (data);
449 /* Got a new socket */
450 if ((pfds->revents & MONO_POLLIN) != 0) {
453 for (i = 1; i < allocated; i++) {
455 if (pfd->fd == -1 || pfd->fd == data->newpfd->fd)
459 if (i == allocated) {
464 allocated = allocated * 2;
465 pfds = g_renew (mono_pollfd, oldfd, allocated);
467 for (; i < allocated; i++)
468 INIT_POLLFD (&pfds [i], -1, 0);
470 #ifndef PLATFORM_WIN32
471 nread = read (data->pipe [0], one, 1);
473 nread = recv ((SOCKET) data->pipe [0], one, 1, 0);
477 return; /* we're closed */
480 INIT_POLLFD (&pfds [i], data->newpfd->fd, data->newpfd->events);
481 ReleaseSemaphore (data->new_sem, 1, NULL);
490 EnterCriticalSection (&data->io_lock);
491 if (data->inited == 0) {
493 LeaveCriticalSection (&data->io_lock);
494 return; /* cleanup called */
497 for (i = 1; i < maxfd && nsock > 0; i++) {
499 if (pfd->fd == -1 || pfd->revents == 0)
503 list = mono_g_hash_table_lookup (data->sock_to_state, GINT_TO_POINTER (pfd->fd));
504 if (list != NULL && (pfd->revents & (MONO_POLLIN | POLL_ERRORS)) != 0) {
505 list = process_io_event (list, MONO_POLLIN);
508 if (list != NULL && (pfd->revents & (MONO_POLLOUT | POLL_ERRORS)) != 0) {
509 list = process_io_event (list, MONO_POLLOUT);
513 mono_g_hash_table_replace (data->sock_to_state, GINT_TO_POINTER (pfd->fd), list);
514 pfd->events = get_events_from_list (list);
516 mono_g_hash_table_remove (data->sock_to_state, GINT_TO_POINTER (pfd->fd));
522 LeaveCriticalSection (&data->io_lock);
527 #define EPOLL_ERRORS (EPOLLERR | EPOLLHUP)
529 socket_io_epoll_main (gpointer p)
534 struct epoll_event *events, *evt;
535 const int nevents = 512;
539 epollfd = data->epollfd;
540 thread = mono_thread_current ();
541 thread->threadpool_thread = TRUE;
542 ves_icall_System_Threading_Thread_SetState (thread, ThreadState_Background);
543 events = g_new0 (struct epoll_event, nevents);
548 if (THREAD_WANTS_A_BREAK (thread))
549 mono_thread_interruption_checkpoint ();
552 g_print ("epoll_wait init\n");
554 ready = epoll_wait (epollfd, events, nevents, -1);
558 g_print ("epoll_wait end with %d ready sockets (%d %s).\n", ready, err, (err) ? g_strerror (err) : "");
562 } while (ready == -1 && errno == EINTR);
568 g_warning ("epoll_wait: %d %s\n", err, g_strerror (err));
574 EnterCriticalSection (&data->io_lock);
575 if (data->inited == 0) {
577 g_print ("data->inited == 0\n");
581 return; /* cleanup called */
584 for (i = 0; i < ready; i++) {
590 list = mono_g_hash_table_lookup (data->sock_to_state, GINT_TO_POINTER (fd));
592 g_print ("Event %d on %d list length: %d\n", evt->events, fd, mono_mlist_length (list));
594 if (list != NULL && (evt->events & (EPOLLIN | EPOLL_ERRORS)) != 0) {
595 list = process_io_event (list, MONO_POLLIN);
598 if (list != NULL && (evt->events & (EPOLLOUT | EPOLL_ERRORS)) != 0) {
599 list = process_io_event (list, MONO_POLLOUT);
603 mono_g_hash_table_replace (data->sock_to_state, GINT_TO_POINTER (fd), list);
604 evt->events = get_events_from_list (list);
606 g_print ("MOD %d to %d\n", fd, evt->events);
608 if (epoll_ctl (epollfd, EPOLL_CTL_MOD, fd, evt)) {
609 if (epoll_ctl (epollfd, EPOLL_CTL_ADD, fd, evt) == -1) {
612 g_message ("epoll_ctl(MOD): %d %s fd: %d events: %d", err, g_strerror (err), fd, evt->events);
618 mono_g_hash_table_remove (data->sock_to_state, GINT_TO_POINTER (fd));
620 g_print ("DEL %d\n", fd);
622 epoll_ctl (epollfd, EPOLL_CTL_DEL, fd, evt);
625 LeaveCriticalSection (&data->io_lock);
631 * select/poll wake up when a socket is closed, but epoll just removes
632 * the socket from its internal list without notification.
635 mono_thread_pool_remove_socket (int sock)
638 MonoMList *list, *next;
639 MonoSocketAsyncResult *state;
641 if (socket_io_data.epoll_disabled == TRUE || socket_io_data.inited == FALSE)
644 EnterCriticalSection (&socket_io_data.io_lock);
645 list = mono_g_hash_table_lookup (socket_io_data.sock_to_state, GINT_TO_POINTER (sock));
647 mono_g_hash_table_remove (socket_io_data.sock_to_state, GINT_TO_POINTER (sock));
649 LeaveCriticalSection (&socket_io_data.io_lock);
652 state = (MonoSocketAsyncResult *) mono_mlist_get_data (list);
653 if (state->operation == AIO_OP_RECEIVE)
654 state->operation = AIO_OP_RECV_JUST_CALLBACK;
655 else if (state->operation == AIO_OP_SEND)
656 state->operation = AIO_OP_SEND_JUST_CALLBACK;
658 next = mono_mlist_remove_item (list, list);
659 list = process_io_event (list, MONO_POLLIN);
661 process_io_event (list, MONO_POLLOUT);
668 #ifdef PLATFORM_WIN32
670 connect_hack (gpointer x)
672 struct sockaddr_in *addr = (struct sockaddr_in *) x;
675 while (connect ((SOCKET) socket_io_data.pipe [1], (SOCKADDR *) addr, sizeof (struct sockaddr_in))) {
678 g_warning ("Error initializing async. sockets %d.\n", WSAGetLastError ());
679 g_assert (WSAGetLastError ());
686 socket_io_init (SocketIOData *data)
688 #ifdef PLATFORM_WIN32
689 struct sockaddr_in server;
690 struct sockaddr_in client;
696 inited = InterlockedCompareExchange (&data->inited, -1, -1);
700 EnterCriticalSection (&data->io_lock);
701 inited = InterlockedCompareExchange (&data->inited, -1, -1);
703 LeaveCriticalSection (&data->io_lock);
708 data->epoll_disabled = (g_getenv ("MONO_DISABLE_AIO") != NULL);
709 if (FALSE == data->epoll_disabled) {
710 data->epollfd = epoll_create (256);
711 data->epoll_disabled = (data->epollfd == -1);
712 if (data->epoll_disabled && g_getenv ("MONO_DEBUG"))
713 g_message ("epoll_create() failed. Using plain poll().");
718 data->epoll_disabled = TRUE;
721 #ifndef PLATFORM_WIN32
722 if (data->epoll_disabled) {
723 if (pipe (data->pipe) != 0) {
733 srv = socket (AF_INET, SOCK_STREAM, IPPROTO_TCP);
734 g_assert (srv != INVALID_SOCKET);
735 data->pipe [1] = socket (AF_INET, SOCK_STREAM, IPPROTO_TCP);
736 g_assert (data->pipe [1] != INVALID_SOCKET);
738 server.sin_family = AF_INET;
739 server.sin_addr.s_addr = inet_addr ("127.0.0.1");
741 if (bind (srv, (SOCKADDR *) &server, sizeof (server))) {
742 g_print ("%d\n", WSAGetLastError ());
746 len = sizeof (server);
747 getsockname (srv, (SOCKADDR *) &server, &len);
749 mono_thread_create (mono_get_root_domain (), connect_hack, &server);
750 len = sizeof (server);
751 data->pipe [0] = accept (srv, (SOCKADDR *) &client, &len);
752 g_assert (data->pipe [0] != INVALID_SOCKET);
755 mono_io_max_worker_threads = mono_max_worker_threads / 2;
756 if (mono_io_max_worker_threads < 10)
757 mono_io_max_worker_threads = 10;
759 data->sock_to_state = mono_g_hash_table_new_type (g_direct_hash, g_direct_equal, MONO_HASH_VALUE_GC);
761 if (data->epoll_disabled) {
762 data->new_sem = CreateSemaphore (NULL, 1, 1, NULL);
763 g_assert (data->new_sem != NULL);
765 io_job_added = CreateSemaphore (NULL, 0, 0x7fffffff, NULL);
766 g_assert (io_job_added != NULL);
767 InitializeCriticalSection (&io_queue_lock);
768 if (data->epoll_disabled) {
769 mono_thread_create (mono_get_root_domain (), socket_io_poll_main, data);
773 mono_thread_create (mono_get_root_domain (), socket_io_epoll_main, data);
776 InterlockedCompareExchange (&data->inited, 1, 0);
777 LeaveCriticalSection (&data->io_lock);
781 socket_io_add_poll (MonoSocketAsyncResult *state)
786 SocketIOData *data = &socket_io_data;
788 #if defined(PLATFORM_MACOSX) || defined(PLATFORM_BSD6) || defined(PLATFORM_WIN32)
789 /* select() for connect() does not work well on the Mac. Bug #75436. */
790 /* Bug #77637 for the BSD 6 case */
791 /* Bug #78888 for the Windows case */
792 if (state->operation == AIO_OP_CONNECT && state->blocking == TRUE) {
793 start_io_thread_or_queue (state);
797 WaitForSingleObject (data->new_sem, INFINITE);
798 if (data->newpfd == NULL)
799 data->newpfd = g_new0 (mono_pollfd, 1);
801 EnterCriticalSection (&data->io_lock);
802 /* FIXME: 64 bit issue: handle can be a pointer on windows? */
803 list = mono_g_hash_table_lookup (data->sock_to_state, GINT_TO_POINTER (state->handle));
805 list = mono_mlist_alloc ((MonoObject*)state);
807 list = mono_mlist_append (list, (MonoObject*)state);
810 events = get_events_from_list (list);
811 INIT_POLLFD (data->newpfd, GPOINTER_TO_INT (state->handle), events);
812 mono_g_hash_table_replace (data->sock_to_state, GINT_TO_POINTER (state->handle), list);
813 LeaveCriticalSection (&data->io_lock);
814 *msg = (char) state->operation;
815 #ifndef PLATFORM_WIN32
816 write (data->pipe [1], msg, 1);
818 send ((SOCKET) data->pipe [1], msg, 1, 0);
824 socket_io_add_epoll (MonoSocketAsyncResult *state)
827 SocketIOData *data = &socket_io_data;
828 struct epoll_event event;
832 memset (&event, 0, sizeof (struct epoll_event));
833 fd = GPOINTER_TO_INT (state->handle);
834 EnterCriticalSection (&data->io_lock);
835 list = mono_g_hash_table_lookup (data->sock_to_state, GINT_TO_POINTER (fd));
837 list = mono_mlist_alloc ((MonoObject*)state);
838 epoll_op = EPOLL_CTL_ADD;
840 list = mono_mlist_append (list, (MonoObject*)state);
841 epoll_op = EPOLL_CTL_MOD;
844 ievt = get_events_from_list (list);
845 if ((ievt & MONO_POLLIN) != 0)
846 event.events |= EPOLLIN;
847 if ((ievt & MONO_POLLOUT) != 0)
848 event.events |= EPOLLOUT;
850 mono_g_hash_table_replace (data->sock_to_state, state->handle, list);
853 g_print ("%s %d with %d\n", epoll_op == EPOLL_CTL_ADD ? "ADD" : "MOD", fd, event.events);
855 if (epoll_ctl (data->epollfd, epoll_op, fd, &event) == -1) {
857 if (epoll_op == EPOLL_CTL_ADD && err == EEXIST) {
858 epoll_op = EPOLL_CTL_MOD;
859 if (epoll_ctl (data->epollfd, epoll_op, fd, &event) == -1) {
860 g_message ("epoll_ctl(MOD): %d %s\n", err, g_strerror (err));
865 LeaveCriticalSection (&data->io_lock);
871 socket_io_add (MonoAsyncResult *ares, MonoSocketAsyncResult *state)
873 socket_io_init (&socket_io_data);
874 MONO_OBJECT_SETREF (state, ares, ares);
876 if (socket_io_data.epoll_disabled == FALSE) {
877 if (socket_io_add_epoll (state))
881 socket_io_add_poll (state);
885 socket_io_filter (MonoObject *target, MonoObject *state)
888 MonoSocketAsyncResult *sock_res = (MonoSocketAsyncResult *) state;
891 if (target == NULL || state == NULL)
894 if (socket_async_call_klass == NULL) {
895 klass = target->vtable->klass;
896 /* Check if it's SocketAsyncCall in System.Net.Sockets
897 * FIXME: check the assembly is signed correctly for extra care
899 if (klass->name [0] == 'S' && strcmp (klass->name, "SocketAsyncCall") == 0
900 && strcmp (mono_image_get_name (klass->image), "System") == 0
901 && klass->nested_in && strcmp (klass->nested_in->name, "Socket") == 0)
902 socket_async_call_klass = klass;
905 if (process_async_call_klass == NULL) {
906 klass = target->vtable->klass;
907 /* Check if it's AsyncReadHandler in System.Diagnostics.Process
908 * FIXME: check the assembly is signed correctly for extra care
910 if (klass->name [0] == 'A' && strcmp (klass->name, "AsyncReadHandler") == 0
911 && strcmp (mono_image_get_name (klass->image), "System") == 0
912 && klass->nested_in && strcmp (klass->nested_in->name, "Process") == 0)
913 process_async_call_klass = klass;
915 /* return both when socket_async_call_klass has not been seen yet and when
916 * the object is not an instance of the class.
918 if (target->vtable->klass != socket_async_call_klass && target->vtable->klass != process_async_call_klass)
921 op = sock_res->operation;
922 if (op < AIO_OP_FIRST || op >= AIO_OP_LAST)
929 mono_async_invoke (MonoAsyncResult *ares)
931 ASyncCall *ac = (ASyncCall *)ares->object_data;
932 MonoThread *thread = NULL;
933 MonoObject *res, *exc = NULL;
934 MonoArray *out_args = NULL;
936 if (ares->execution_context) {
937 /* use captured ExecutionContext (if available) */
938 thread = mono_thread_current ();
939 MONO_OBJECT_SETREF (ares, original_context, thread->execution_context);
940 MONO_OBJECT_SETREF (thread, execution_context, ares->execution_context);
942 ares->original_context = NULL;
946 res = mono_message_invoke (ares->async_delegate, ac->msg, &exc, &out_args);
947 MONO_OBJECT_SETREF (ac, res, res);
948 MONO_OBJECT_SETREF (ac, msg->exc, exc);
949 MONO_OBJECT_SETREF (ac, out_args, out_args);
953 /* call async callback if cb_method != null*/
955 MonoObject *exc = NULL;
957 mono_runtime_invoke (ac->cb_method, ac->cb_target, pa, &exc);
958 /* 'exc' will be the previous ac->msg->exc if not NULL and not
959 * catched. If catched, this will be set to NULL and the
960 * exception will not be printed. */
961 MONO_OBJECT_SETREF (ac->msg, exc, exc);
964 /* restore original thread execution context if flow isn't suppressed, i.e. non null */
965 if (ares->original_context) {
966 MONO_OBJECT_SETREF (thread, execution_context, ares->original_context);
967 ares->original_context = NULL;
970 /* notify listeners */
971 mono_monitor_enter ((MonoObject *) ares);
972 if (ares->handle != NULL) {
973 ac->wait_event = (gsize) mono_wait_handle_get_handle ((MonoWaitHandle *) ares->handle);
974 SetEvent ((gpointer)(gsize)ac->wait_event);
976 mono_monitor_exit ((MonoObject *) ares);
978 EnterCriticalSection (&ares_lock);
979 mono_g_hash_table_remove (ares_htable, ares);
980 LeaveCriticalSection (&ares_lock);
984 mono_thread_pool_init ()
987 int threads_per_cpu = THREADS_PER_CPU;
989 if ((int) InterlockedCompareExchange (&tp_inited, 1, 0) == 1)
992 MONO_GC_REGISTER_ROOT (ares_htable);
993 MONO_GC_REGISTER_ROOT (socket_io_data.sock_to_state);
994 InitializeCriticalSection (&socket_io_data.io_lock);
995 InitializeCriticalSection (&ares_lock);
996 ares_htable = mono_g_hash_table_new_type (NULL, NULL, MONO_HASH_KEY_VALUE_GC);
997 job_added = CreateSemaphore (NULL, 0, 0x7fffffff, NULL);
998 g_assert (job_added != NULL);
999 GetSystemInfo (&info);
1000 if (g_getenv ("MONO_THREADS_PER_CPU") != NULL) {
1001 threads_per_cpu = atoi (g_getenv ("MONO_THREADS_PER_CPU"));
1002 if (threads_per_cpu <= 0)
1003 threads_per_cpu = THREADS_PER_CPU;
1006 mono_max_worker_threads = 20 + threads_per_cpu * info.dwNumberOfProcessors;
1008 async_call_klass = mono_class_from_name (mono_defaults.corlib, "System", "MonoAsyncCall");
1009 g_assert (async_call_klass);
1013 mono_thread_pool_add (MonoObject *target, MonoMethodMessage *msg, MonoDelegate *async_callback,
1016 MonoDomain *domain = mono_domain_get ();
1017 MonoAsyncResult *ares;
1020 ac = (ASyncCall*)mono_object_new (mono_domain_get (), async_call_klass);
1021 MONO_OBJECT_SETREF (ac, msg, msg);
1022 MONO_OBJECT_SETREF (ac, state, state);
1024 if (async_callback) {
1025 ac->cb_method = mono_get_delegate_invoke (((MonoObject *)async_callback)->vtable->klass);
1026 MONO_OBJECT_SETREF (ac, cb_target, async_callback);
1029 ares = mono_async_result_new (domain, NULL, ac->state, NULL, (MonoObject*)ac);
1030 MONO_OBJECT_SETREF (ares, async_delegate, target);
1032 EnterCriticalSection (&ares_lock);
1033 mono_g_hash_table_insert (ares_htable, ares, ares);
1034 LeaveCriticalSection (&ares_lock);
1036 if (socket_io_filter (target, state)) {
1037 socket_io_add (ares, (MonoSocketAsyncResult *) state);
1041 start_thread_or_queue (ares);
1046 start_thread_or_queue (MonoAsyncResult *ares)
1050 busy = (int) InterlockedCompareExchange (&busy_worker_threads, 0, -1);
1051 worker = (int) InterlockedCompareExchange (&mono_worker_threads, 0, -1);
1052 if (worker <= ++busy &&
1053 worker < mono_max_worker_threads) {
1054 InterlockedIncrement (&mono_worker_threads);
1055 InterlockedIncrement (&busy_worker_threads);
1056 mono_thread_create (mono_get_root_domain (), async_invoke_thread, ares);
1058 append_job (&mono_delegate_section, &async_call_queue, (MonoObject*)ares);
1059 ReleaseSemaphore (job_added, 1, NULL);
1064 mono_thread_pool_finish (MonoAsyncResult *ares, MonoArray **out_args, MonoObject **exc)
1071 /* check if already finished */
1072 mono_monitor_enter ((MonoObject *) ares);
1074 if (ares->endinvoke_called) {
1075 *exc = (MonoObject *)mono_exception_from_name (mono_defaults.corlib, "System",
1076 "InvalidOperationException");
1077 mono_monitor_exit ((MonoObject *) ares);
1081 ares->endinvoke_called = 1;
1082 ac = (ASyncCall *)ares->object_data;
1084 g_assert (ac != NULL);
1086 /* wait until we are really finished */
1087 if (!ares->completed) {
1088 if (ares->handle == NULL) {
1089 ac->wait_event = (gsize)CreateEvent (NULL, TRUE, FALSE, NULL);
1090 g_assert(ac->wait_event != 0);
1091 MONO_OBJECT_SETREF (ares, handle, (MonoObject *) mono_wait_handle_new (mono_object_domain (ares), (gpointer)(gsize)ac->wait_event));
1093 mono_monitor_exit ((MonoObject *) ares);
1094 WaitForSingleObjectEx ((gpointer)(gsize)ac->wait_event, INFINITE, TRUE);
1096 mono_monitor_exit ((MonoObject *) ares);
1099 *exc = ac->msg->exc; /* FIXME: GC add write barrier */
1100 *out_args = ac->out_args;
1106 mono_thread_pool_cleanup (void)
1110 EnterCriticalSection (&mono_delegate_section);
1111 free_queue (&async_call_queue);
1112 release = (gint) InterlockedCompareExchange (&mono_worker_threads, 0, -1);
1113 LeaveCriticalSection (&mono_delegate_section);
1115 ReleaseSemaphore (job_added, release, NULL);
1117 socket_io_cleanup (&socket_io_data);
1121 append_job (CRITICAL_SECTION *cs, TPQueue *list, MonoObject *ar)
1123 EnterCriticalSection (cs);
1124 if (list->array && (list->next_elem < mono_array_length (list->array))) {
1125 mono_array_setref (list->array, list->next_elem, ar);
1127 LeaveCriticalSection (cs);
1131 MONO_GC_REGISTER_ROOT (list->array);
1132 list->array = mono_array_new (mono_get_root_domain (), mono_defaults.object_class, 16);
1134 int count = list->next_elem - list->first_elem;
1135 /* slide the array or create a larger one if it's full */
1136 if (list->first_elem) {
1137 mono_array_memcpy_refs (list->array, 0, list->array, list->first_elem, count);
1139 MonoArray *newa = mono_array_new (mono_get_root_domain (), mono_defaults.object_class, mono_array_length (list->array) * 2);
1140 mono_array_memcpy_refs (newa, 0, list->array, list->first_elem, count);
1143 list->first_elem = 0;
1144 list->next_elem = count;
1146 mono_array_setref (list->array, list->next_elem, ar);
1148 LeaveCriticalSection (cs);
1152 dequeue_job (CRITICAL_SECTION *cs, TPQueue *list)
1157 EnterCriticalSection (cs);
1158 if (!list->array || list->first_elem == list->next_elem) {
1159 LeaveCriticalSection (cs);
1162 ar = mono_array_get (list->array, MonoObject*, list->first_elem);
1164 count = list->next_elem - list->first_elem;
1165 /* reduce the size of the array if it's mostly empty */
1166 if (mono_array_length (list->array) > 16 && count < (mono_array_length (list->array) / 3)) {
1167 MonoArray *newa = mono_array_new (mono_get_root_domain (), mono_defaults.object_class, mono_array_length (list->array) / 2);
1168 mono_array_memcpy_refs (newa, 0, list->array, list->first_elem, count);
1170 list->first_elem = 0;
1171 list->next_elem = count;
1173 LeaveCriticalSection (cs);
1179 free_queue (TPQueue *list)
1182 list->first_elem = list->next_elem = 0;
1186 async_invoke_thread (gpointer data)
1192 thread = mono_thread_current ();
1193 thread->threadpool_thread = TRUE;
1194 ves_icall_System_Threading_Thread_SetState (thread, ThreadState_Background);
1197 MonoAsyncResult *ar;
1199 ar = (MonoAsyncResult *) data;
1201 /* worker threads invokes methods in different domains,
1202 * so we need to set the right domain here */
1203 domain = ((MonoObject *)ar)->vtable->domain;
1204 mono_thread_push_appdomain_ref (domain);
1205 if (mono_domain_set (domain, FALSE)) {
1208 mono_async_invoke (ar);
1209 ac = (ASyncCall *) ar->object_data;
1211 if (ac->msg->exc != NULL)
1212 mono_unhandled_exception (ac->msg->exc);
1214 mono_domain_set (mono_get_root_domain (), TRUE);
1216 mono_thread_pop_appdomain_ref ();
1217 InterlockedDecrement (&busy_worker_threads);
1220 data = dequeue_job (&mono_delegate_section, &async_call_queue);
1224 int timeout = 10000;
1225 guint32 start_time = GetTickCount ();
1228 wr = WaitForSingleObjectEx (job_added, (guint32)timeout, TRUE);
1229 if (THREAD_WANTS_A_BREAK (thread))
1230 mono_thread_interruption_checkpoint ();
1232 timeout -= GetTickCount () - start_time;
1234 if (wr != WAIT_TIMEOUT)
1235 data = dequeue_job (&mono_delegate_section, &async_call_queue);
1237 while (!data && timeout > 0);
1241 workers = (int) InterlockedCompareExchange (&mono_worker_threads, 0, -1);
1242 min = (int) InterlockedCompareExchange (&mono_min_worker_threads, 0, -1);
1244 while (!data && workers <= min) {
1245 WaitForSingleObjectEx (job_added, INFINITE, TRUE);
1246 if (THREAD_WANTS_A_BREAK (thread))
1247 mono_thread_interruption_checkpoint ();
1249 data = dequeue_job (&mono_delegate_section, &async_call_queue);
1250 workers = (int) InterlockedCompareExchange (&mono_worker_threads, 0, -1);
1251 min = (int) InterlockedCompareExchange (&mono_min_worker_threads, 0, -1);
1256 InterlockedDecrement (&mono_worker_threads);
1260 InterlockedIncrement (&busy_worker_threads);
1263 g_assert_not_reached ();
1267 ves_icall_System_Threading_ThreadPool_GetAvailableThreads (gint *workerThreads, gint *completionPortThreads)
1271 MONO_ARCH_SAVE_REGS;
1273 busy = (gint) InterlockedCompareExchange (&busy_worker_threads, 0, -1);
1274 busy_io = (gint) InterlockedCompareExchange (&busy_io_worker_threads, 0, -1);
1275 *workerThreads = mono_max_worker_threads - busy;
1276 *completionPortThreads = mono_io_max_worker_threads - busy_io;
1280 ves_icall_System_Threading_ThreadPool_GetMaxThreads (gint *workerThreads, gint *completionPortThreads)
1282 MONO_ARCH_SAVE_REGS;
1284 *workerThreads = mono_max_worker_threads;
1285 *completionPortThreads = mono_io_max_worker_threads;
1289 ves_icall_System_Threading_ThreadPool_GetMinThreads (gint *workerThreads, gint *completionPortThreads)
1291 gint workers, workers_io;
1293 MONO_ARCH_SAVE_REGS;
1295 workers = (gint) InterlockedCompareExchange (&mono_min_worker_threads, 0, -1);
1296 workers_io = (gint) InterlockedCompareExchange (&mono_io_min_worker_threads, 0, -1);
1298 *workerThreads = workers;
1299 *completionPortThreads = workers_io;
1303 ves_icall_System_Threading_ThreadPool_SetMinThreads (gint workerThreads, gint completionPortThreads)
1305 MONO_ARCH_SAVE_REGS;
1307 if (workerThreads < 0 || workerThreads > mono_max_worker_threads)
1310 if (completionPortThreads < 0 || completionPortThreads > mono_io_max_worker_threads)
1313 InterlockedExchange (&mono_min_worker_threads, workerThreads);
1314 InterlockedExchange (&mono_io_min_worker_threads, completionPortThreads);
1315 /* FIXME: should actually start the idle threads if needed */
1320 ves_icall_System_Threading_ThreadPool_SetMaxThreads (gint workerThreads, gint completionPortThreads)
1322 MONO_ARCH_SAVE_REGS;
1324 if (workerThreads < mono_max_worker_threads)
1327 if (completionPortThreads < mono_io_max_worker_threads)
1330 InterlockedExchange (&mono_max_worker_threads, workerThreads);
1331 InterlockedExchange (&mono_io_max_worker_threads, completionPortThreads);